// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <turbo/log/logging.h>
#include <gtest/gtest.h>
#include <cstdint>
#include <functional>
#include <limits>
#include <optional>

#include <pollux/testing/gtest_utils.h>
#include <pollux/common/memory/byte_stream.h>
#include <pollux/common/testutil/optional_empty.h>
#include <pollux/serializers/presto_serializer.h>
#include <pollux/vector/base_vector.h>
#include <pollux/vector/complex_vector.h>
#include <pollux/vector/decoded_vector.h>
#include <pollux/vector/dictionary_vector.h>
#include <pollux/vector/flat_vector.h>
#include <pollux/vector/lazy_vector.h>
#include <pollux/vector/selectivity_vector.h>
#include <pollux/vector/simple_vector.h>
#include <pollux/vector/type_aliases.h>
#include <pollux/vector/vector_type_utils.h>
#include <pollux/testing/vector/vector_test_base.h>

namespace kumo::pollux {
    namespace {
        // LazyVector loader for testing. Minimal implementation that documents the API
        // contract.
        class TestingLoader : public VectorLoader {
        public:
            explicit TestingLoader(VectorPtr data) : data_(data), rowCounter_(0) {
            }

            void loadInternal(
                RowSet rows,
                ValueHook *hook,
                vector_size_t resultSize,
                VectorPtr *result) override {
                if (hook) {
                    POLLUX_DYNAMIC_SCALAR_TYPE_DISPATCH(
                        applyHook, data_->type_kind(), rows, hook);
                    return;
                }
                *result = data_;
                POLLUX_CHECK_GE(data_->size(), resultSize);
                rowCounter_ += rows.size();
            }

            int32_t rowCount() const {
                return rowCounter_;
            }

        private:
            template<TypeKind Kind>
            void applyHook(RowSet rows, ValueHook *hook) {
                using T = typename TypeTraits<Kind>::NativeType;
                auto values = data_->as<SimpleVector<T> >();
                bool acceptsNulls = hook->acceptsNulls();
                for (auto i = 0; i < rows.size(); ++i) {
                    auto row = rows[i];
                    if (values->is_null_at(row)) {
                        if (acceptsNulls) {
                            hook->addNull(i);
                        }
                    } else {
                        T value = values->value_at(row);
                        if constexpr (std::is_same_v<T, Timestamp>) {
                            POLLUX_FAIL();
                        } else {
                            hook->addValueTyped(i, value);
                        }
                    }
                }
            }

            VectorPtr data_;
            int32_t rowCounter_;
        };

        struct NonPOD {
            static int alive;

            int x;

            NonPOD(int x = 123) : x(x) {
                ++alive;
            }

            ~NonPOD() {
                --alive;
            }

            bool operator==(const NonPOD &other) const {
                return x == other.x;
            }
        };

        int NonPOD::alive = 0;

        class VectorTest : public testing::Test, public pollux::VectorBuilder {
        protected:
            static void SetUpTestCase() {
                memory::MemoryManager::testingSetInstance({});
            }

            void SetUp() override {
                if (!isRegisteredVectorSerde()) {
                    kumo::pollux::serializer::presto::PrestoVectorSerde::
                            registerVectorSerde();
                }
            }

            template<typename T>
            T testValue(int32_t i, BufferPtr &space) {
                return i;
            }

            template<TypeKind KIND>
            VectorPtr createScalar(TypePtr type, vector_size_t size, bool withNulls) {
                using T = typename TypeTraits<KIND>::NativeType;
                BufferPtr buffer;
                VectorPtr base = BaseVector::create(type, size, pool());
                auto flat = std::dynamic_pointer_cast<FlatVector<T> >(base);
                for (int32_t i = 0; i < flat->size(); ++i) {
                    if (withNulls && i % 3 == 0) {
                        flat->set_null(i, true);
                    } else {
                        flat->set(i, testValue<T>(i, buffer));
                    }
                }
                return base;
            }

            VectorPtr createRow(int32_t numRows, bool withNulls) {
                auto childType =
                        ROW({"child_bigint", "child_string"}, {BIGINT(), VARCHAR()});
                auto parentType =
                        ROW({"parent_bigint", "parent_row"}, {BIGINT(), childType});
                auto baseRow = BaseVector::create(parentType, numRows, pool());
                auto row = baseRow->as<RowVector>();
                EXPECT_EQ(row->size(), numRows);
                EXPECT_EQ(row->nulls(), nullptr);

                std::vector<VectorPtr> nested = {
                    createScalar<TypeKind::BIGINT>(BIGINT(), numRows, withNulls),
                    createScalar<TypeKind::VARCHAR>(VARCHAR(), numRows, withNulls)
                };
                auto childRow = std::make_shared<RowVector>(
                    pool(),
                    childType,
                    BufferPtr(nullptr),
                    numRows,
                    std::move(nested),
                    0 /*nullCount*/);
                BufferPtr nulls;
                if (withNulls) {
                    nulls = allocate_nulls(numRows, pool());
                    auto rawNulls = nulls->asMutable<uint64_t>();
                    for (int32_t i = 0; i < numRows; ++i) {
                        bits::set_null(rawNulls, i, i % 8 == 0);
                    }
                }
                std::vector<VectorPtr> parentFields = {
                    createScalar<TypeKind::BIGINT>(BIGINT(), numRows, withNulls), childRow
                };
                return std::make_shared<RowVector>(
                    pool(),
                    parentType,
                    nulls,
                    numRows,
                    std::move(parentFields),
                    BaseVector::count_nulls(nulls, numRows));
            }

            int32_t createRepeated(
                int32_t numRows,
                bool withNulls,
                BufferPtr *nulls,
                BufferPtr *offsets,
                BufferPtr *sizes,
                int forceWidth) {
                *offsets = AlignedBuffer::allocate<vector_size_t>(numRows, pool());
                auto *rawOffsets = (*offsets)->asMutable<vector_size_t>();

                *sizes = AlignedBuffer::allocate<vector_size_t>(numRows, pool());
                auto *rawSizes = (*sizes)->asMutable<vector_size_t>();

                uint64_t *rawNulls = nullptr;
                if (withNulls) {
                    *nulls = allocate_nulls(numRows, pool());
                    rawNulls = (*nulls)->asMutable<uint64_t>();
                }
                int32_t offset = 0;
                for (int32_t i = 0; i < numRows; ++i) {
                    int32_t size = (forceWidth > 0) ? forceWidth : i % 7;
                    if (withNulls && i % 5 == 0) {
                        bits::set_null(rawNulls, i, true);
                        if (forceWidth == 0) {
                            // in forceWidth case, the size of a null element is still
                            // the same as a non-null element.
                            rawOffsets[i] = 0;
                            rawSizes[i] = 0;
                            continue;
                        }
                    }
                    rawOffsets[i] = offset;
                    rawSizes[i] = size;
                    offset += size;
                }
                return offset;
            }

            VectorPtr createArray(int32_t numRows, bool withNulls) {
                BufferPtr nulls;
                BufferPtr offsets;
                BufferPtr sizes;
                int32_t numElements =
                        createRepeated(numRows, withNulls, &nulls, &offsets, &sizes, 0);
                VectorPtr elements = createRow(numElements, withNulls);
                return std::make_shared<ArrayVector>(
                    pool(),
                    ARRAY(elements->type()),
                    nulls,
                    numRows,
                    offsets,
                    sizes,
                    elements,
                    BaseVector::count_nulls(nulls, numRows));
            }

            VectorPtr createMap(int32_t numRows, bool withNulls);

            template<TypeKind KIND>
            void testFlat(TypePtr type, vector_size_t size) {
                testFlat<KIND>(type, size, false);
                testFlat<KIND>(type, size, true);
            }

            template<TypeKind KIND>
            void testFlat(TypePtr type, vector_size_t size, bool withNulls) {
                using T = typename TypeTraits<KIND>::NativeType;
                VectorPtr base = BaseVector::create(type, size, pool());
                auto flat = std::dynamic_pointer_cast<FlatVector<T> >(base);
                ASSERT_NE(flat.get(), nullptr);
                EXPECT_EQ(flat->size(), size);
                EXPECT_GE(flat->values()->size(), BaseVector::byteSize<T>(size));
                EXPECT_EQ(flat->nulls(), nullptr);

                flat->resize(size * 2);
                EXPECT_EQ(flat->size(), size * 2);
                EXPECT_GE(flat->values()->capacity(), BaseVector::byteSize<T>(size * 2));
                EXPECT_EQ(flat->nulls(), nullptr);

                if (withNulls) {
                    flat->set_null(size * 2 - 1, true);
                    EXPECT_EQ(flat->raw_nulls()[0], bits::kNotNull64);
                    EXPECT_GE(
                        flat->nulls()->capacity(), BaseVector::byteSize<bool>(flat->size()));
                    EXPECT_TRUE(flat->is_null_at(size * 2 - 1));
                }

                // Test that the null is cleared.
                BufferPtr buffer;
                flat->set(size * 2 - 1, testValue<T>(size, buffer));
                EXPECT_FALSE(flat->is_null_at(size * 2 - 1));

                if (withNulls) {
                    // Check that new elements are initialized as not null after downsize
                    // and upsize.
                    flat->set_null(size * 2 - 1, true);
                    flat->resize(size * 2 - 1);
                    flat->resize(size * 2);
                    EXPECT_FALSE(flat->is_null_at(size * 2 - 1));
                }

                // Check that new StringView elements are initialized as empty after
                // downsize and upsize which does not involve a capacity change.
                if constexpr (std::is_same_v<T, StringView>) {
                    flat->mutableRawValues()[size * 2 - 1] = StringView("a");
                    flat->resize(size * 2 - 1);
                    flat->resize(size * 2);
                    EXPECT_EQ(flat->value_at(size * 2 - 1).size(), 0);
                }

                // Fill, the values at size * 2 - 1 gets assigned a second time.
                for (int32_t i = 0; i < flat->size(); ++i) {
                    if (withNulls && i % 3 == 0) {
                        flat->set_null(i, true);
                    } else {
                        flat->set(i, testValue<T>(i, buffer));
                    }
                }

                for (int32_t i = 0; i < flat->size(); ++i) {
                    if (withNulls && i % 3 == 0) {
                        EXPECT_TRUE(flat->is_null_at(i));
                    } else {
                        EXPECT_FALSE(flat->is_null_at(i));
                        if constexpr (KIND == TypeKind::OPAQUE) {
                            EXPECT_TRUE(flat->value_at(i));
                            EXPECT_EQ(
                                *std::static_pointer_cast<NonPOD>(flat->value_at(i)),
                                *std::static_pointer_cast<NonPOD>(testValue<T>(i, buffer)));
                        } else {
                            EXPECT_EQ(flat->value_at(i), testValue<T>(i, buffer));
                        }
                    }
                }

                testCopy(flat, numIterations_);
                testSlices(flat);

                // Check that type kind is preserved in cases where T is the same with.
                EXPECT_EQ(
                    flat->type_kind(), BaseVector::wrap_in_constant(100, 0, flat)->type_kind());
            }

            static SelectivityVector selectEven(vector_size_t size) {
                SelectivityVector even(size);
                for (auto i = 0; i < size; ++i) {
                    even.setValid(i, i % 2 == 0);
                }
                even.updateBounds();
                return even;
            }

            static SelectivityVector selectOdd(vector_size_t size) {
                SelectivityVector odd(size);
                for (auto i = 0; i < size; ++i) {
                    odd.setValid(i, i % 2 != 0);
                }
                odd.updateBounds();
                return odd;
            }

            static std::string printEncodings(const VectorPtr &vector) {
                std::stringstream out;
                out << vector->encoding();

                VectorPtr inner = vector;
                do {
                    switch (inner->encoding()) {
                        case VectorEncoding::Simple::DICTIONARY:
                        case VectorEncoding::Simple::CONSTANT:
                            inner = inner->value_vector();
                            if (inner == nullptr) {
                                return out.str();
                            }
                            break;
                        default:
                            return out.str();
                    }

                    out << ", " << inner->encoding();
                } while (true);
                POLLUX_UNREACHABLE();
            }

            void testCopyEncoded(VectorPtr source) {
                bool maybeConstant = false;
                auto sourcePtr = source.get();
                for (;;) {
                    auto encoding = sourcePtr->encoding();
                    maybeConstant = encoding == VectorEncoding::Simple::CONSTANT ||
                                    encoding == VectorEncoding::Simple::LAZY;
                    if (maybeConstant) {
                        break;
                    }
                    if (encoding != VectorEncoding::Simple::DICTIONARY) {
                        break;
                    }
                    sourcePtr = sourcePtr->value_vector().get();
                }

                auto kind = source->type_kind();
                auto isSmall = kind == TypeKind::BOOLEAN || kind == TypeKind::TINYINT;
                auto sourceSize = source->size();
                auto target = BaseVector::create(source->type(), sourceSize, pool());
                // Writes target out of sequence by copying the first half of
                // source to odd positions and the second half to even positions.
                auto even = selectEven(sourceSize);
                auto odd = selectOdd(sourceSize);
                std::vector<vector_size_t> evenSource(sourceSize, INT32_MAX);
                std::vector<vector_size_t> oddSource(sourceSize, INT32_MAX);

                for (auto i = 0; i < sourceSize; ++i) {
                    if (i % 2 == 0) {
                        evenSource[i] = i / 2;
                    } else {
                        oddSource[i] = (sourceSize / 2) + (i / 2);
                    }
                }
                target->copy(source.get(), even, &evenSource[0]);
                target->copy(source.get(), odd, &oddSource[0]);
                // We check that a wrapped source tests the same via equal_value_at
                // on the wrapping and via the translation of DecodedVector.
                SelectivityVector allRows(sourceSize);
                DecodedVector decoded(*source, allRows);
                auto base = decoded.base();
                auto nulls = decoded.nulls(&allRows);
                auto indices = decoded.indices();
                for (int32_t i = 0; i < sourceSize; ++i) {
                    if (i % 2 == 0) {
                        auto sourceIdx = evenSource[i];
                        EXPECT_TRUE(target->equal_value_at(source.get(), i, sourceIdx))
            << "at " << i << ": " << target->toString(i) << " vs. "
            << source->toString(sourceIdx);

                        // We check the same with 'decoded'.
                        if (source->is_null_at(sourceIdx)) {
                            EXPECT_TRUE(decoded.is_null_at(sourceIdx));
                            EXPECT_TRUE(nulls && bits::isBitNull(nulls, sourceIdx));
                            EXPECT_TRUE(target->is_null_at(i));
                        } else {
                            EXPECT_FALSE(nulls && bits::isBitNull(nulls, sourceIdx));
                            EXPECT_FALSE(decoded.is_null_at(sourceIdx));
                            EXPECT_FALSE(target->is_null_at(i));
                            EXPECT_TRUE(target->equal_value_at(
                                base, i, indices ? indices[sourceIdx] : sourceIdx));
                        }
                    } else {
                        EXPECT_TRUE(target->equal_value_at(source.get(), i, oddSource[i]));
                        // We check the same with 'decoded'.
                        auto sourceIdx = oddSource[i];
                        if (source->is_null_at(sourceIdx)) {
                            EXPECT_TRUE(nulls && bits::isBitNull(nulls, sourceIdx));
                            EXPECT_TRUE(target->is_null_at(i));
                        } else {
                            EXPECT_FALSE(nulls && bits::isBitNull(nulls, sourceIdx));
                            EXPECT_FALSE(target->is_null_at(i));
                            EXPECT_TRUE(target->equal_value_at(
                                base, i, indices ? indices[sourceIdx] : sourceIdx));
                        }
                    }
                    if (i > 1 && i < sourceSize - 1 && !target->is_null_at(i) && !isSmall &&
                        !maybeConstant && source->is_scalar()) {
                        EXPECT_FALSE(target->equal_value_at(source.get(), i, i));
                    }
                }

                target->resize(target->size() * 2);
                // Nulls must be clear after resize.
                for (int32_t i = target->size() / 2; i < target->size(); ++i) {
                    EXPECT_FALSE(target->is_null_at(i));
                }
                // Copy all of source to second half of target.
                target->copy(source.get(), sourceSize, 0, sourceSize);
                for (int32_t i = 0; i < sourceSize; ++i) {
                    EXPECT_TRUE(target->equal_value_at(source.get(), sourceSize + i, i));
                    EXPECT_TRUE(source->equal_value_at(target.get(), i, sourceSize + i));
                }

                std::vector<BaseVector::CopyRange> ranges = {
                    {0, 0, sourceSize},
                    {0, sourceSize, sourceSize},
                };
                target->copy_ranges(source.get(), ranges);
                for (int32_t i = 0; i < sourceSize; ++i) {
                    EXPECT_TRUE(source->equal_value_at(target.get(), i, i));
                    EXPECT_TRUE(source->equal_value_at(target.get(), i, sourceSize + i));
                }

                // Check that uninitialized is copyable.
                target->resize(target->size() + 100);
                target->copy(target.get(), target->size() - 50, target->size() - 100, 50);
            }

            /// Add up to 'level' wrappings to 'source' and test copy and serialize
            /// operations.
            void testCopy(VectorPtr source, int level) {
                SCOPED_TRACE(printEncodings(source));
                testCopyEncoded(source);

                if (source->type()->is_opaque()) {
                    // No support for serialization of opaque types yet, make sure something
                    // throws
                    EXPECT_THROW(testSerialization(source), std::exception);
                } else {
                    testSerialization(source);
                }

                if (level == 0) {
                    return;
                }

                // Add dictionary wrapping to put vectors elements in reverse order.
                // [1, 2, 3, 4] becomes [4, 3, 2, 1].
                // If 'source' has nulls, add more nulls every 11-th row starting with row #
                // 'level'.
                auto sourceSize = source->size();
                BufferPtr dictionaryNulls;
                uint64_t *rawNulls = nullptr;
                if (source->may_have_nulls()) {
                    dictionaryNulls = allocate_nulls(sourceSize, pool());
                    rawNulls = dictionaryNulls->asMutable<uint64_t>();
                }
                BufferPtr indices =
                        AlignedBuffer::allocate<vector_size_t>(sourceSize, pool());
                for (int32_t i = 0; i < sourceSize; ++i) {
                    indices->asMutable<vector_size_t>()[i] = sourceSize - i - 1;
                    if (rawNulls && (i + level) % 11 == 0) {
                        bits::set_null(rawNulls, i);
                    }
                }
                auto inDictionary = BaseVector::wrap_in_dictionary(
                    dictionaryNulls, indices, sourceSize, source);
                testCopy(inDictionary, level - 1);

                // Add constant wrapping.
                auto constant = BaseVector::wrap_in_constant(20, 10 + level, source);
                testCopy(constant, level - 1);

                // Add constant wrapping using null row.
                if (source->may_have_nulls() && !source->is_lazy()) {
                    int32_t firstNull = 0;
                    for (; firstNull < sourceSize; ++firstNull) {
                        if (source->is_null_at(firstNull)) {
                            break;
                        }
                    }
                    constant = BaseVector::wrap_in_constant(firstNull + 123, firstNull, source);
                    testCopy(constant, level - 1);
                }

                // Add lazy wrapping.
                auto lazy = std::make_shared<LazyVector>(
                    source->pool(),
                    source->type(),
                    sourceSize,
                    std::make_unique<TestingLoader>(source));
                testCopy(lazy, level - 1);
            }

            void testCopyFromAllNulls(
                const VectorPtr &vector,
                const VectorPtr &allNullSource) {
                SCOPED_TRACE(vector->toString());

                const vector_size_t size = 1'000;

                ASSERT_GE(vector->size(), size);

                // Save a copy of the 'vector' to compare results after copy.
                auto vectorCopy = BaseVector::copy(*vector);

                // Copy every 3-rd row.
                SelectivityVector rowsToCopy(size, false);
                for (auto i = 0; i < size; i += 3) {
                    rowsToCopy.setValid(i, true);
                }
                rowsToCopy.updateBounds();

                SelectivityVector rowsToKeep(size);
                rowsToKeep.deselect(rowsToCopy);

                // Copy from row N to N - 10 for N >= 0 and from row N to N for N < 10;
                std::vector<vector_size_t> toSourceRow(size);
                for (auto i = 0; i < size; i += 3) {
                    if (i < 10) {
                        toSourceRow[i] = i;
                    } else {
                        toSourceRow[i] = i - 10;
                    }
                }

                vector->copy(allNullSource.get(), rowsToCopy, toSourceRow.data());

                rowsToCopy.applyToSelected(
                    [&](auto row) { EXPECT_TRUE(vector->is_null_at(row)) << "at " << row; });

                rowsToKeep.applyToSelected([&](vector_size_t row) {
                    EXPECT_FALSE(vector->is_null_at(row));
                    EXPECT_TRUE(vector->equal_value_at(vectorCopy.get(), row, row))
          << "at " << row << ": " << vector->toString(row) << " vs. "
          << vectorCopy->toString(row);
                });
            }

            void testCopySingleRangeFromAllNulls(
                const VectorPtr &vector,
                const VectorPtr &allNullSource) {
                SCOPED_TRACE(vector->toString());

                const vector_size_t size = 1'000;

                ASSERT_GE(vector->size(), size);

                // Save a copy of the 'vector' to compare results after copy.
                auto vectorCopy = BaseVector::copy(*vector);

                vector->copy(allNullSource.get(), 40, 33, 78);

                for (auto i = 0; i < size; ++i) {
                    if (i < 40 || i >= 40 + 78) {
                        EXPECT_FALSE(vector->is_null_at(i));
                        EXPECT_TRUE(vector->equal_value_at(vectorCopy.get(), i, i))
            << "at " << i << ": " << vector->toString(i) << " vs. "
            << vectorCopy->toString(i);
                    } else {
                        EXPECT_TRUE(vector->is_null_at(i)) << "at " << i;
                    }
                }
            }

            void testCopyRangesFromAllNulls(
                const VectorPtr &vector,
                const VectorPtr &allNullSource) {
                SCOPED_TRACE(vector->toString());

                const vector_size_t size = 1'000;

                ASSERT_GE(vector->size(), size);

                // Save a copy of the 'vector' to compare results after copy.
                auto vectorCopy = BaseVector::copy(*vector);

                std::vector<BaseVector::CopyRange> rangesToCopy = {
                    {0, 0, 7},
                    {10, 12, 5},
                    {100, 500, 79},
                    {200, 601, 1},
                    {990, 950, 10},
                };

                std::vector<BaseVector::CopyRange> rangesToKeep = {
                    {0, 7, 5},
                    {0, 17, 500 - 17},
                    {0, 579, 601 - 579},
                    {0, 602, 950 - 602},
                    {0, 960, 40},
                };

                vector->copy_ranges(allNullSource.get(), rangesToCopy);

                for (const auto &range: rangesToCopy) {
                    for (auto i = 0; i < range.count; ++i) {
                        EXPECT_TRUE(vector->is_null_at(range.targetIndex + i));
                    }
                }

                for (const auto &range: rangesToKeep) {
                    for (auto i = 0; i < range.count; ++i) {
                        auto index = range.targetIndex + i;
                        EXPECT_FALSE(vector->is_null_at(index));
                        EXPECT_TRUE(vector->equal_value_at(vectorCopy.get(), index, index))
            << "at " << index << ": " << vector->toString(index) << " vs. "
            << vectorCopy->toString(index);
                    }
                }
            }

            static void testSlice(
                const VectorPtr &vec,
                int level,
                vector_size_t offset,
                vector_size_t length) {
                SCOPED_TRACE(fmt::format(
                    "testSlice encoding={} offset={} length={}",
                    vec->encoding(),
                    offset,
                    length));
                ASSERT_GE(vec->size(), offset + length);
                auto slice = vec->loaded_vector()->slice(offset, length);
                ASSERT_EQ(slice->size(), length);
                // Compare values and nulls directly.
                for (int i = 0; i < length; ++i) {
                    EXPECT_TRUE(slice->equal_value_at(vec.get(), i, offset + i));
                } {
                    // Check DecodedVector works on slice.
                    SelectivityVector allRows(length);
                    DecodedVector decoded(*slice, allRows);
                    auto base = decoded.base();
                    auto indices = decoded.indices();
                    for (int i = 0; i < length; ++i) {
                        auto j = offset + i;
                        if (vec->is_null_at(j)) {
                            EXPECT_TRUE(decoded.is_null_at(i));
                        } else {
                            ASSERT_FALSE(decoded.is_null_at(i));
                            auto ii = indices ? indices[i] : i;
                            EXPECT_TRUE(base->equal_value_at(vec.get(), ii, j));
                        }
                    }
                }
                if (level == 0) {
                    return;
                }
                // Add constant wrapping.
                if (auto i = 3 + level; i < slice->size()) {
                    testSlices(BaseVector::wrap_in_constant(123, i, slice), level - 1);
                }
                // Add constant wrapping using null row.
                if (slice->may_have_nulls() && !slice->is_lazy()) {
                    for (int i = 0; i < slice->size(); ++i) {
                        if (slice->is_null_at(i)) {
                            testSlices(BaseVector::wrap_in_constant(i + 123, i, slice), level - 1);
                            break;
                        }
                    }
                } {
                    // Add dictionary wrapping to put vectors elements in reverse order.  If
                    // 'source' has nulls, add more nulls every 11-th row starting with row #
                    // 'level'.
                    BufferPtr nulls;
                    uint64_t *rawNulls = nullptr;
                    if (slice->may_have_nulls()) {
                        nulls = allocate_nulls(slice->size(), slice->pool());
                        rawNulls = nulls->asMutable<uint64_t>();
                    }
                    BufferPtr indices =
                            AlignedBuffer::allocate<vector_size_t>(slice->size(), slice->pool());
                    auto rawIndices = indices->asMutable<vector_size_t>();
                    for (vector_size_t i = 0; i < slice->size(); ++i) {
                        rawIndices[i] = slice->size() - i - 1;
                        if (rawNulls && (i + level) % 11 == 0) {
                            bits::set_null(rawNulls, i);
                        }
                    }
                    auto wrapped =
                            BaseVector::wrap_in_dictionary(nulls, indices, slice->size(), slice);
                    testSlices(wrapped, level - 1);
                } {
                    // Add lazy wrapping.
                    auto wrapped = std::make_shared<LazyVector>(
                        slice->pool(),
                        slice->type(),
                        slice->size(),
                        std::make_unique<TestingLoader>(slice));
                    testSlices(wrapped, level - 1);
                } {
                    // Test that resize works even when the underlying buffers are
                    // immutable. This is true for a slice since it creates buffer views over
                    // the buffers of the original vector that it sliced.
                    auto newSize = slice->size() * 2;
                    slice->resize(newSize);
                    EXPECT_EQ(slice->size(), newSize);
                }
            }

            static void testSlices(const VectorPtr &slice, int level = 2) {
                for (vector_size_t offset: {0, 16, 17}) {
                    for (vector_size_t length: {0, 1, 83}) {
                        if (offset + length <= slice->size()) {
                            testSlice(slice, level, offset, length);
                        }
                    }
                }
            }

            std::unique_ptr<ByteInputStream> prepareInput(std::string &string) {
                // Put 'string' in 'input' in many pieces.
                const int32_t size = string.size();
                std::vector<ByteRange> ranges;
                for (int32_t i = 0; i < 10; ++i) {
                    int32_t start = i * (size / 10);
                    int32_t end = (i == 9) ? size : (i + 1) * (size / 10);
                    ranges.emplace_back();
                    ranges.back().buffer = reinterpret_cast<uint8_t *>(&string[start]);
                    ranges.back().size = end - start;
                    ranges.back().position = 0;
                }

                return std::make_unique<BufferInputStream>(std::move(ranges));
            }

            void checkSizes(
                BaseVector *source,
                std::vector<vector_size_t> &sizes,
                std::string &string) {
                int32_t total = 0;
                for (auto size: sizes) {
                    total += size;
                }
                // Approx. 10 bytes for encoding, 4 for length, 1 for null presence. This
                // can vary.
                total -= 15;
                float tolerance = source->may_have_nulls() ? 0.3 : 0.2;
                switch (source->type_kind()) {
                    case TypeKind::BOOLEAN:
                        tolerance = 10;
                        break;
                    case TypeKind::TINYINT:
                        tolerance = 2;
                        break;
                    case TypeKind::SMALLINT:
                        tolerance = 1.5;
                        break;
                    case TypeKind::TIMESTAMP:
                        // The size is reported as 16 bytes but Presto wire format is 8 bytes.
                        tolerance = 2.2;
                        break;
                    default:
                        break;
                }
                if (total > 1024) {
                    int32_t size = string.size();
                    EXPECT_GE(total, size * (1 - tolerance));
                    EXPECT_LE(total, size * (1 + tolerance));
                }
            }

            /// Serialize odd and even rows into separate streams. Read serialized data
            /// back and compare with the original data.
            void testSerialization(VectorPtr source) {
                // Serialization functions expect loaded vectors.
                source = BaseVector::loaded_vector_shared(source);
                auto sourceRow = make_row_vector({"c"}, {source});
                auto sourceRowType = as_row_type(sourceRow->type());

                VectorStreamGroup even(pool(), nullptr);
                even.createStreamTree(sourceRowType, source->size() / 4);

                VectorStreamGroup odd(pool(), nullptr);
                odd.createStreamTree(sourceRowType, source->size() / 3);

                std::vector<IndexRange> evenIndices;
                std::vector<IndexRange> oddIndices;
                for (vector_size_t i = 0; i < source->size(); ++i) {
                    if (i % 2 == 0) {
                        evenIndices.push_back(IndexRange{i, 1});
                    } else {
                        oddIndices.push_back(IndexRange{i, 1});
                    }
                }

                std::vector<vector_size_t> evenSizes(evenIndices.size());
                std::vector<vector_size_t *> evenSizePointers(evenSizes.size());
                for (int32_t i = 0; i < evenSizes.size(); ++i) {
                    evenSizePointers[i] = &evenSizes[i];
                }

                std::vector<vector_size_t> oddSizes(oddIndices.size());
                std::vector<vector_size_t *> oddSizePointers(oddSizes.size());
                for (int32_t i = 0; i < oddSizes.size(); ++i) {
                    oddSizePointers[i] = &oddSizes[i];
                }

                VectorStreamGroup::estimateSerializedSize(
                    source.get(), evenIndices, nullptr, evenSizePointers.data());
                VectorStreamGroup::estimateSerializedSize(
                    source.get(), oddIndices, nullptr, oddSizePointers.data());
                even.append(
                    sourceRow, melon::Range(evenIndices.data(), evenIndices.size() / 2));
                even.append(
                    sourceRow,
                    melon::Range(
                        &evenIndices[evenIndices.size() / 2],
                        evenIndices.size() - evenIndices.size() / 2));
                odd.append(
                    sourceRow, melon::Range(oddIndices.data(), oddIndices.size() / 2));
                odd.append(
                    sourceRow,
                    melon::Range(
                        &oddIndices[oddIndices.size() / 2],
                        oddIndices.size() - oddIndices.size() / 2));

                std::stringstream evenStream;
                std::stringstream oddStream;
                OStreamOutputStream evenOutputStream(&evenStream);
                OStreamOutputStream oddOutputStream(&oddStream);
                even.flush(&evenOutputStream);
                odd.flush(&oddOutputStream);

                auto evenString = evenStream.str();
                checkSizes(source.get(), evenSizes, evenString);

                auto evenInput = prepareInput(evenString);

                RowVectorPtr resultRow;
                VectorStreamGroup::read(
                    evenInput.get(), pool(), sourceRowType, nullptr, &resultRow, nullptr);
                VectorPtr result = resultRow->childAt(0);
                switch (source->encoding()) {
                    case VectorEncoding::Simple::FLAT:
                    case VectorEncoding::Simple::ROW:
                    case VectorEncoding::Simple::ARRAY:
                    case VectorEncoding::Simple::MAP: {
                        // Test that retained sizes are in reasonable bounds if vectors
                        // are not wrapped.
                        auto originalRetained = source->retained_size();
                        auto deserializedRetained = resultRow->retained_size();
                        EXPECT_LE(deserializedRetained, originalRetained)
            << "Deserializing half is larger than original";
                        EXPECT_GT(deserializedRetained, 0);
                        break;
                    }
                    default:
                        // skip non-flat encodings
                        break;
                }

                for (int32_t i = 0; i < evenIndices.size(); ++i) {
                    EXPECT_TRUE(result->equal_value_at(source.get(), i, evenIndices[i].begin))
          << "at " << i << ", " << source->encoding();
                }

                auto oddString = oddStream.str();
                auto oddInput = prepareInput(oddString);

                VectorStreamGroup::read(
                    oddInput.get(), pool(), sourceRowType, nullptr, &resultRow, nullptr);
                result = resultRow->childAt(0);
                for (int32_t i = 0; i < oddIndices.size(); ++i) {
                    EXPECT_TRUE(result->equal_value_at(source.get(), i, oddIndices[i].begin))
          << "at " << i << ", " << source->encoding();
                }
            }

            template<typename T>
            static void testArrayOrMapSliceMutability(const std::shared_ptr<T> &vec) {
                ASSERT_GE(vec->size(), 3);
                auto slice = std::dynamic_pointer_cast<T>(vec->slice(0, 3));
                auto offsets = slice->rawOffsets();
                slice->mutableOffsets(slice->size());
                EXPECT_NE(slice->rawOffsets(), offsets);
                auto sizes = slice->rawSizes();
                slice->mutableSizes(slice->size());
                EXPECT_NE(slice->rawSizes(), sizes);
            }

            size_t vectorSize_{100};
            size_t numIterations_{3};
        };

        template<>
        int128_t VectorTest::testValue<int128_t>(int32_t i, BufferPtr & /*space*/) {
            return HugeInt::build(i % 2 ? (i * -1) : i, 0xAAAAAAAAAAAAAAAA);
        }

        template<>
        StringView VectorTest::testValue(int32_t n, BufferPtr &buffer) {
            if (!buffer || buffer->capacity() < 1000) {
                buffer = AlignedBuffer::allocate<char>(1000, pool());
            }
            std::stringstream out;
            out << n;
            for (int32_t i = 0; i < n % 20; ++i) {
                out << " " << i * i;
            }
            std::string str = out.str();
            EXPECT_LE(str.size(), buffer->capacity());
            memcpy(buffer->asMutable<char>(), str.data(), str.size());
            return StringView(buffer->as<char>(), str.size());
        }

        template<>
        bool VectorTest::testValue(int32_t i, BufferPtr & /*space*/) {
            return (i % 2) == 1;
        }

        template<>
        Timestamp VectorTest::testValue(int32_t i, BufferPtr & /*space*/) {
            // Return even milliseconds.
            return Timestamp(i * 1000, (i % 1000) * 1000000);
        }

        template<>
        std::shared_ptr<void> VectorTest::testValue(int32_t i, BufferPtr & /*space*/) {
            return std::make_shared<NonPOD>(i);
        }

        VectorPtr VectorTest::createMap(int32_t numRows, bool withNulls) {
            BufferPtr nulls;
            BufferPtr offsets;
            BufferPtr sizes;
            int32_t numElements =
                    createRepeated(numRows, withNulls, &nulls, &offsets, &sizes, 0);
            VectorPtr elements = createRow(numElements, withNulls);
            auto keysBase = BaseVector::create(VARCHAR(), 7, pool());
            auto flatKeys = keysBase->as<FlatVector<StringView> >();
            for (int32_t i = 0; i < keysBase->size(); ++i) {
                BufferPtr buffer;
                flatKeys->set(i, testValue<StringView>(1000 + i, buffer));
            }

            auto indices =
                    AlignedBuffer::allocate<vector_size_t>(elements->size(), pool());

            auto rawSizes = sizes->as<vector_size_t>();
            int32_t offset = 0;
            for (int32_t i = 0; i < numRows; ++i) {
                int32_t size = rawSizes[i];
                for (int32_t index = 0; index < size; ++index) {
                    indices->asMutable<vector_size_t>()[offset++] = index;
                }
            }
            POLLUX_CHECK_EQ(offset, elements->size());
            auto keys = BaseVector::wrap_in_dictionary(
                BufferPtr(nullptr),
                std::move(indices),
                elements->size(),
                std::move(keysBase));
            return std::make_shared<MapVector>(
                pool(),
                MAP(VARCHAR(), elements->type()),
                nulls,
                numRows,
                offsets,
                sizes,
                keys,
                elements,
                BaseVector::count_nulls(nulls, numRows));
        }

        TEST_F(VectorTest, createInt) {
            testFlat<TypeKind::BIGINT>(BIGINT(), vectorSize_);
            testFlat<TypeKind::INTEGER>(INTEGER(), vectorSize_);
            testFlat<TypeKind::SMALLINT>(SMALLINT(), vectorSize_);
            testFlat<TypeKind::TINYINT>(TINYINT(), vectorSize_);
        }

        TEST_F(VectorTest, createDouble) {
            testFlat<TypeKind::REAL>(REAL(), vectorSize_);
            testFlat<TypeKind::DOUBLE>(DOUBLE(), vectorSize_);
        }

        TEST_F(VectorTest, createStr) {
            testFlat<TypeKind::VARCHAR>(VARCHAR(), vectorSize_);
            testFlat<TypeKind::VARBINARY>(VARBINARY(), vectorSize_);
        }

        TEST_F(VectorTest, createOther) {
            testFlat<TypeKind::BOOLEAN>(BOOLEAN(), vectorSize_);
            testFlat<TypeKind::TIMESTAMP>(TIMESTAMP(), vectorSize_);
            testFlat<TypeKind::INTEGER>(DATE(), vectorSize_);
            testFlat<TypeKind::BIGINT>(INTERVAL_DAY_TIME(), vectorSize_);
        }

        TEST_F(VectorTest, createDecimal) {
            testFlat<TypeKind::BIGINT>(DECIMAL(10, 5), vectorSize_);
            testFlat<TypeKind::HUGEINT>(DECIMAL(30, 5), vectorSize_);
        }

        TEST_F(VectorTest, createOpaque) {
            NonPOD::alive = 0;
            testFlat<TypeKind::OPAQUE>(OPAQUE<int>(), vectorSize_);
            EXPECT_EQ(NonPOD::alive, 0);
        }

        TEST_F(VectorTest, get_or_create_empty) {
            auto empty = BaseVector::get_or_create_empty(nullptr, VARCHAR(), pool());
            EXPECT_NE(empty, nullptr);
            EXPECT_EQ(empty->size(), 0);
            EXPECT_EQ(empty->type(), VARCHAR());
        }

        TEST_F(VectorTest, row) {
            auto baseRow = createRow(vectorSize_, false);
            testCopy(baseRow, numIterations_);
            testSlices(baseRow);
            baseRow = createRow(vectorSize_, true);
            testCopy(baseRow, numIterations_);
            testSlices(baseRow);
            auto allNull = BaseVector::create_null_constant(baseRow->type(), 50, pool());
            testCopy(allNull, numIterations_);
            testSlices(allNull);
            // created from BaseVector::Create()
            baseRow = BaseVector::create(baseRow->type(), vectorSize_, pool());
            testCopy(baseRow, numIterations_);
            testSlices(baseRow);

            // Check child accessors throws.
            auto row_vector = baseRow->as<RowVector>();
            EXPECT_NO_THROW(row_vector->childAt(1)->size());
            EXPECT_NO_THROW(row_vector->childAt(1)->resize(vectorSize_));

            EXPECT_THROW(row_vector->childAt(3)->size(), PolluxRuntimeError);
            POLLUX_ASSERT_THROW(
                row_vector->childAt(3)->size(),
                "Trying to access non-existing child in RowVector:");

            EXPECT_THROW(row_vector->childAt(3)->resize(vectorSize_), PolluxRuntimeError);
            POLLUX_ASSERT_THROW(
                row_vector->childAt(3)->resize(vectorSize_),
                "Trying to access non-existing child in RowVector:");

            ASSERT_EQ(
                row_vector->childAt("parent_bigint").get(), row_vector->childAt(0).get());
            ASSERT_EQ(
                row_vector->childAt("parent_row").get(), row_vector->childAt(1).get());
            POLLUX_ASSERT_THROW(
                row_vector->childAt("foo"),
                "Field not found: foo. Available fields are: parent_bigint, parent_row.");
        }

        TEST_F(VectorTest, array) {
            auto baseArray = createArray(vectorSize_, false);
            testCopy(baseArray, numIterations_);
            testSlices(baseArray);
            baseArray = createArray(vectorSize_, true);
            testCopy(baseArray, numIterations_);
            testSlices(baseArray);
            auto allNull = BaseVector::create_null_constant(baseArray->type(), 50, pool());
            testCopy(allNull, numIterations_);
            testSlices(allNull);
        }

        TEST_F(VectorTest, map) {
            auto baseMap = createMap(vectorSize_, false);
            testCopy(baseMap, numIterations_);
            testSlices(baseMap);
            baseMap = createRow(vectorSize_, true);
            testCopy(baseMap, numIterations_);
            testSlices(baseMap);
            auto allNull = BaseVector::create_null_constant(baseMap->type(), 50, pool());
            testCopy(allNull, numIterations_);
            testSlices(allNull);
        }

        TEST_F(VectorTest, unknown) {
            // Creates a const UNKNOWN vector.
            auto constUnknownVector = BaseVector::create_constant(
                UNKNOWN(), variant(TypeKind::UNKNOWN), 123, pool());
            ASSERT_FALSE(constUnknownVector->is_scalar());
            ASSERT_EQ(TypeKind::UNKNOWN, constUnknownVector->type_kind());
            ASSERT_EQ(123, constUnknownVector->size());
            for (auto i = 0; i < constUnknownVector->size(); i++) {
                ASSERT_TRUE(constUnknownVector->is_null_at(i));
            }

            // Create an int vector and copy UNKNOWN const vector into it.
            auto intVector = BaseVector::create(BIGINT(), 10, pool());
            ASSERT_FALSE(intVector->is_null_at(0));
            ASSERT_FALSE(intVector->may_have_nulls());
            intVector->copy(constUnknownVector.get(), 0, 0, 1);
            ASSERT_TRUE(intVector->is_null_at(0));

            SelectivityVector rows(2);
            intVector->copy(constUnknownVector.get(), rows, nullptr);
            for (int i = 0; i < intVector->size(); ++i) {
                if (i < 2) {
                    ASSERT_TRUE(intVector->is_null_at(i));
                } else {
                    ASSERT_FALSE(intVector->is_null_at(i));
                }
            }

            // Can't copy to constant.
            EXPECT_ANY_THROW(constUnknownVector->copy(intVector.get(), rows, nullptr));

            // Create a flat UNKNOWN vector.
            auto unknownVector = BaseVector::create(UNKNOWN(), 10, pool());
            ASSERT_EQ(VectorEncoding::Simple::FLAT, unknownVector->encoding());
            ASSERT_FALSE(unknownVector->is_scalar());
            for (int i = 0; i < unknownVector->size(); ++i) {
                ASSERT_TRUE(unknownVector->is_null_at(i)) << i;
            }
            // Can't copy to UNKNOWN vector.
            EXPECT_ANY_THROW(unknownVector->copy(intVector.get(), rows, nullptr));

            // Copying flat UNKNOWN into integer vector nulls out all the copied element.
            ASSERT_FALSE(intVector->is_null_at(3));
            intVector->copy(unknownVector.get(), 3, 3, 1);
            ASSERT_TRUE(intVector->is_null_at(3));

            rows.resize(intVector->size());
            intVector->copy(unknownVector.get(), rows, nullptr);
            for (int i = 0; i < intVector->size(); ++i) {
                ASSERT_TRUE(intVector->is_null_at(i));
            }

            // It is okay to copy to a non-constant UNKNOWN vector.
            unknownVector->copy(constUnknownVector.get(), rows, nullptr);
            for (int i = 0; i < unknownVector->size(); ++i) {
                ASSERT_TRUE(unknownVector->is_null_at(i)) << i;
            }

            // Can't copy to constant.
            EXPECT_ANY_THROW(
                constUnknownVector->copy(unknownVector.get(), rows, nullptr));

            for (auto &vec: {constUnknownVector, unknownVector}) {
                auto slice = vec->slice(1, 3);
                ASSERT_EQ(TypeKind::UNKNOWN, slice->type_kind());
                ASSERT_EQ(3, slice->size());
                for (auto i = 0; i < slice->size(); i++) {
                    ASSERT_TRUE(slice->is_null_at(i));
                }
            }
        }

        TEST_F(VectorTest, copyBoolAllNullFlatVector) {
            const vector_size_t size = 1'000;
            auto allNulls = make_all_null_flat_vector<bool>(size);

            auto copy = make_flat_vector<bool>(size, [](auto row) { return row % 2 == 0; });

            // Copy first 10 rows from allNulls.
            SelectivityVector rows(10);
            copy->copy(allNulls.get(), rows, nullptr);

            for (auto i = 0; i < size; i++) {
                ASSERT_EQ(copy->is_null_at(i), i < 10) << "at " << i;
                if (i >= 10) {
                    ASSERT_EQ(copy->value_at(i), i % 2 == 0) << "at " << i;
                }
            }
        }

        /// Test copying non-null values into an all-nulls flat vector.
        TEST_F(VectorTest, copyToAllNullsFlatVector) {
            // Create all-nulls flat vector with values buffer unset.
            const vector_size_t size = 1'000;
            auto allNulls = make_all_null_flat_vector<int32_t>(size);

            auto source = make_flat_vector<int32_t>({0, 1, 2, 3, 4});

            allNulls->copy(source.get(), 0, 0, 3);

            for (auto i = 0; i < 3; ++i) {
                ASSERT_EQ(i, allNulls->value_at(i));
            }
            for (auto i = 3; i < size; ++i) {
                ASSERT_TRUE(allNulls->is_null_at(i));
            }

            // Reset allNulls vector back to all nulls.
            allNulls = make_all_null_flat_vector<int32_t>(size);

            SelectivityVector rows(4);
            allNulls->copy(source.get(), rows, nullptr);

            for (auto i = 0; i < 4; ++i) {
                ASSERT_EQ(i, allNulls->value_at(i));
            }
            for (auto i = 4; i < size; ++i) {
                ASSERT_TRUE(allNulls->is_null_at(i));
            }
        }

        template<TypeKind kind>
        static VectorPtr createAllNullsFlatVector(
            vector_size_t size,
            memory::MemoryPool *pool,
            const TypePtr &type) {
            using T = typename TypeTraits<kind>::NativeType;

            return std::make_shared<FlatVector<T> >(
                pool,
                type,
                allocate_nulls(size, pool, bits::kNull),
                size,
                nullptr,
                std::vector<BufferPtr>());
        }

        VectorPtr createAllNullsVector(
            const TypePtr &type,
            vector_size_t size,
            memory::MemoryPool *pool) {
            auto kind = type->kind();
            switch (kind) {
                case TypeKind::ROW: {
                    std::vector<VectorPtr> children(type->size(), nullptr);
                    return std::make_shared<RowVector>(
                        pool, type, allocate_nulls(size, pool, bits::kNull), size, children);
                }
                case TypeKind::ARRAY:
                    return std::make_shared<ArrayVector>(
                        pool,
                        type,
                        allocate_nulls(size, pool, bits::kNull),
                        size,
                        allocateSizes(size, pool),
                        allocateSizes(size, pool),
                        nullptr);
                case TypeKind::MAP:
                    return std::make_shared<MapVector>(
                        pool,
                        type,
                        allocate_nulls(size, pool, bits::kNull),
                        size,
                        allocateSizes(size, pool),
                        allocateSizes(size, pool),
                        nullptr,
                        nullptr);
                default:
                    return POLLUX_DYNAMIC_SCALAR_TYPE_DISPATCH(
                                createAllNullsFlatVector, kind, size, pool, type);
            }
        }

        TEST_F(VectorTest, copyFromAllNulls) {
            vector_size_t size = 1'000;

            auto test = [&](const auto &makeVectorFunc) {
                auto vector = makeVectorFunc();
                auto allNullSource =
                        createAllNullsVector(vector->type(), vector->size(), pool());

                testCopyFromAllNulls(vector, allNullSource);

                vector = makeVectorFunc();
                testCopySingleRangeFromAllNulls(vector, allNullSource);

                vector = makeVectorFunc();
                testCopyRangesFromAllNulls(vector, allNullSource);
            };

            // Copy to BIGINT.
            test([&]() {
                return make_flat_vector<int64_t>(size, [](auto row) { return row; });
            });

            // Copy to BOOLEAN.
            test([&]() {
                return make_flat_vector<bool>(size, [](auto row) { return row % 7 == 3; });
            });

            // Copy to VARCHAR.
            test([&]() {
                return make_flat_vector<std::string>(
                    size, [](auto row) { return std::string(row % 17, 'x'); });
            });

            // Copy to ARRAY.
            test([&]() {
                return make_array_vector<int64_t>(
                    size, [](auto row) { return row % 7; }, [](auto row) { return row; });
            });

            // Copy to MAP.
            test([&]() {
                return make_map_vector<int64_t, double>(
                    size,
                    [](auto row) { return row % 7; },
                    [](auto row) { return row; },
                    [](auto row) { return row * 0.1; });
            });

            // TODO Enable after fixing
            // https://github.com/facebookincubator/pollux/issues/6612
            //  // Copy to ROW.
            //  test([&]() {
            //    return make_row_vector({
            //        make_flat_vector<int64_t>(size, [](auto row) { return row; }),
            //        make_flat_vector<double>(size, [](auto row) { return row * 0.1; }),
            //    });
            //  });
        }

        TEST_F(VectorTest, copyFromUnknown) {
            vector_size_t size = 1'000;
            auto unknown = make_all_null_flat_vector<UnknownValue>(size);

            auto test = [&](const auto &makeVectorFunc) {
                auto vector = makeVectorFunc();
                testCopyFromAllNulls(vector, unknown);

                vector = makeVectorFunc();
                testCopySingleRangeFromAllNulls(vector, unknown);

                vector = makeVectorFunc();
                testCopyRangesFromAllNulls(vector, unknown);
            };

            // Copy to BIGINT.
            test([&]() {
                return make_flat_vector<int64_t>(size, [](auto row) { return row; });
            });

            // Copy to BOOLEAN.
            test([&]() {
                return make_flat_vector<bool>(size, [](auto row) { return row % 7 == 3; });
            });

            // Copy to VARCHAR.
            test([&]() {
                return make_flat_vector<std::string>(
                    size, [](auto row) { return std::string(row % 17, 'x'); });
            });

            // Copy to ARRAY.
            test([&]() {
                return make_array_vector<int64_t>(
                    size, [](auto row) { return row % 7; }, [](auto row) { return row; });
            });

            // Copy to MAP.
            test([&]() {
                return make_map_vector<int64_t, double>(
                    size,
                    [](auto row) { return row % 7; },
                    [](auto row) { return row; },
                    [](auto row) { return row * 0.1; });
            });

            // Copy to ROW.
            test([&]() {
                return make_row_vector({
                    make_flat_vector<int64_t>(size, [](auto row) { return row; }),
                    make_flat_vector<double>(size, [](auto row) { return row * 0.1; }),
                });
            });
        }

        TEST_F(VectorTest, wrap_in_constant) {
            // wrap flat vector
            const vector_size_t size = 1'000;
            auto flat_vector =
                    make_flat_vector<int32_t>(size, [](auto row) { return row; }, null_every(7));

            auto constVector = std::dynamic_pointer_cast<ConstantVector<int32_t> >(
                BaseVector::wrap_in_constant(size, 5, flat_vector));
            EXPECT_EQ(constVector->value_vector(), nullptr);
            for (auto i = 0; i < size; i++) {
                ASSERT_FALSE(constVector->is_null_at(i));
                ASSERT_EQ(5, constVector->value_at(i));
            }

            constVector = std::dynamic_pointer_cast<ConstantVector<int32_t> >(
                BaseVector::wrap_in_constant(size, 7, flat_vector));
            EXPECT_EQ(constVector->value_vector(), nullptr);
            for (auto i = 0; i < size; i++) {
                ASSERT_TRUE(constVector->is_null_at(i));
            }

            // wrap dictionary vector
            BufferPtr indices = AlignedBuffer::allocate<vector_size_t>(size, pool());
            auto rawIndices = indices->asMutable<vector_size_t>();
            for (auto i = 0; i < size; i++) {
                rawIndices[i] = 2 * i % size;
            }

            BufferPtr nulls = allocate_nulls(size, pool());
            auto rawNulls = nulls->asMutable<uint64_t>();
            for (auto i = 0; i < size; i++) {
                bits::set_null(rawNulls, i, i % 11 == 0);
            }

            auto dictVector =
                    BaseVector::wrap_in_dictionary(nulls, indices, size, flat_vector);

            constVector = std::dynamic_pointer_cast<ConstantVector<int32_t> >(
                BaseVector::wrap_in_constant(size, 5, dictVector));
            EXPECT_EQ(constVector->value_vector(), nullptr);
            for (auto i = 0; i < size; i++) {
                ASSERT_FALSE(constVector->is_null_at(i));
                ASSERT_EQ(10, constVector->value_at(i));
            }

            constVector = std::dynamic_pointer_cast<ConstantVector<int32_t> >(
                BaseVector::wrap_in_constant(size, 11, dictVector));
            EXPECT_EQ(constVector->value_vector(), nullptr);
            for (auto i = 0; i < size; i++) {
                ASSERT_TRUE(constVector->is_null_at(i));
            }

            // Wrap constant with value_vector.
            auto array_vector = make_array_vector<int32_t>(
                size, [](auto) { return 10; }, [](auto i) { return i; }, null_every(7));
            auto constBaseVector = std::make_shared<ConstantVector<ComplexType> >(
                pool(), size, 3, array_vector);
            auto constArrayVector =
                    std::dynamic_pointer_cast<ConstantVector<ComplexType> >(
                        BaseVector::wrap_in_constant(size, 22, constBaseVector));
            EXPECT_NE(constArrayVector->value_vector(), nullptr);
            EXPECT_EQ(constArrayVector->value_vector(), array_vector);
            for (auto i = 0; i < size; i++) {
                ASSERT_FALSE(constArrayVector->is_null_at(i));
                ASSERT_TRUE(constArrayVector->equal_value_at(array_vector.get(), i, 3));
            }

            // Wrap constant with value_vector and null value.
            constBaseVector = std::make_shared<ConstantVector<ComplexType> >(
                pool(), size, 7, array_vector);
            constArrayVector = std::dynamic_pointer_cast<ConstantVector<ComplexType> >(
                BaseVector::wrap_in_constant(size, 22, constBaseVector, true));
            EXPECT_NE(constArrayVector->value_vector(), nullptr);
            for (auto i = 0; i < size; i++) {
                ASSERT_TRUE(constArrayVector->is_null_at(i));
            }

            // Wrap a loaded lazy complex vector that will be retained as a value_vector.
            // Ensure the lazy layer is stripped away and the value_vector points to the
            // loaded Vector underneath it.
            auto lazyOverArray = std::make_shared<LazyVector>(
                pool(),
                array_vector->type(),
                size,
                std::make_unique<TestingLoader>(array_vector));
            lazyOverArray->loaded_vector();
            EXPECT_TRUE(lazyOverArray->isLoaded());
            constArrayVector = std::dynamic_pointer_cast<ConstantVector<ComplexType> >(
                BaseVector::wrap_in_constant(size, 22, lazyOverArray));
            EXPECT_FALSE(constArrayVector->value_vector()->is_lazy());
        }

        TEST_F(VectorTest, wrapInConstantWithCopy) {
            // Wrap flat vector.
            const vector_size_t size = 1'000;
            auto flat_vector =
                    make_flat_vector<int32_t>(size, [](auto row) { return row; }, null_every(7));

            auto constVector = std::dynamic_pointer_cast<ConstantVector<int32_t> >(
                BaseVector::wrap_in_constant(size, 5, flat_vector, true));
            EXPECT_EQ(constVector->value_vector(), nullptr);
            for (auto i = 0; i < size; i++) {
                ASSERT_FALSE(constVector->is_null_at(i));
                ASSERT_EQ(5, constVector->value_at(i));
            }

            constVector = std::dynamic_pointer_cast<ConstantVector<int32_t> >(
                BaseVector::wrap_in_constant(size, 7, flat_vector, true));
            EXPECT_EQ(constVector->value_vector(), nullptr);
            for (auto i = 0; i < size; i++) {
                ASSERT_TRUE(constVector->is_null_at(i));
            }

            // Wrap dictionary vector.
            BufferPtr indices = AlignedBuffer::allocate<vector_size_t>(size, pool());
            auto rawIndices = indices->asMutable<vector_size_t>();
            for (auto i = 0; i < size; i++) {
                rawIndices[i] = 2 * i % size;
            }

            BufferPtr nulls = allocate_nulls(size, pool());
            auto rawNulls = nulls->asMutable<uint64_t>();
            for (auto i = 0; i < size; i++) {
                bits::set_null(rawNulls, i, i % 11 == 0);
            }

            auto dictVector =
                    BaseVector::wrap_in_dictionary(nulls, indices, size, flat_vector);

            constVector = std::dynamic_pointer_cast<ConstantVector<int32_t> >(
                BaseVector::wrap_in_constant(size, 5, dictVector, true));
            EXPECT_EQ(constVector->value_vector(), nullptr);
            for (auto i = 0; i < size; i++) {
                ASSERT_FALSE(constVector->is_null_at(i));
                ASSERT_EQ(10, constVector->value_at(i));
            }

            // Wrap constant with value_vector.
            auto array_vector = make_array_vector<int32_t>(
                size, [](auto) { return 10; }, [](auto i) { return i; }, null_every(7));
            auto constBaseVector = std::make_shared<ConstantVector<ComplexType> >(
                pool(), size, 3, array_vector);
            auto constArrayVector =
                    std::dynamic_pointer_cast<ConstantVector<ComplexType> >(
                        BaseVector::wrap_in_constant(size, 22, constBaseVector, true));
            EXPECT_NE(constArrayVector->value_vector(), nullptr);
            EXPECT_EQ(constArrayVector->value_vector().use_count(), 1);
            for (auto i = 0; i < size; i++) {
                ASSERT_FALSE(constArrayVector->is_null_at(i));
                ASSERT_TRUE(constArrayVector->equal_value_at(array_vector.get(), i, 3));
            }

            // Wrap constant with value_vector and null value.
            constBaseVector = std::make_shared<ConstantVector<ComplexType> >(
                pool(), size, 7, array_vector);
            constArrayVector = std::dynamic_pointer_cast<ConstantVector<ComplexType> >(
                BaseVector::wrap_in_constant(size, 22, constBaseVector, true));
            EXPECT_NE(constArrayVector->value_vector(), nullptr);
            EXPECT_EQ(constArrayVector->value_vector().use_count(), 1);
            for (auto i = 0; i < size; i++) {
                ASSERT_TRUE(constArrayVector->is_null_at(i));
            }
        }

        TEST_F(VectorTest, rowResize) {
            auto testRowResize = [&](const VectorPtr &vector, bool setNotNull) {
                auto row_vector = vector->as<RowVector>();
                for (auto &child: row_vector->children()) {
                    POLLUX_CHECK_EQ(child.use_count(), 1);
                }
                auto oldSize = row_vector->size();
                auto newSize = oldSize * 2;

                row_vector->resize(newSize, setNotNull);

                EXPECT_EQ(row_vector->size(), newSize);

                for (auto &child: row_vector->children()) {
                    EXPECT_EQ(child->size(), newSize);
                }

                if (setNotNull) {
                    for (int i = oldSize; i < newSize; i++) {
                        EXPECT_EQ(row_vector->is_null_at(i), !setNotNull);
                        for (auto &child: row_vector->children()) {
                            EXPECT_EQ(child->is_null_at(i), !setNotNull);
                        }
                    }
                }
            };

            // FlatVectors.
            auto row_vector =
                    make_row_vector({make_flat_vector<int32_t>(10), make_flat_vector<int64_t>(10)});
            testRowResize(row_vector, bits::kNull);

            row_vector =
                    make_row_vector({make_flat_vector<int32_t>(10), make_flat_vector<double>(10)});
            testRowResize(row_vector, bits::kNotNull);

            row_vector = make_row_vector({make_flat_vector<StringView>(10)});
            testRowResize(row_vector, bits::kNotNull);
            row_vector = make_row_vector({make_flat_vector<StringView>(10)});
            testRowResize(row_vector, bits::kNull);

            // Dictionaries.
            row_vector = make_row_vector({
                BaseVector::wrap_in_dictionary(
                    nullptr,
                    make_indices(10, [](auto row) { return row; }),
                    10,
                    BaseVector::wrap_in_dictionary(
                        nullptr,
                        make_indices(10, [](auto row) { return row; }),
                        10,
                        make_flat_vector<int32_t>(10)))
            });
            testRowResize(row_vector, bits::kNotNull);

            // Constants.
            row_vector = make_row_vector({
                BaseVector::wrap_in_constant(
                    10,
                    5,
                    make_array_vector<int32_t>(
                        10,
                        [](auto row) { return row % 5 + 1; },
                        [](auto row, auto index) { return row * 2 + index; }))
            });
            testRowResize(row_vector, bits::kNotNull);

            // Complex Types.
            row_vector = make_row_vector(
                {
                    make_array_vector<int32_t>(
                        10,
                        [](auto row) { return row % 5 + 1; },
                        [](auto row, auto index) { return row * 2 + index; }),
                    make_map_vector<int32_t, int64_t>(
                        10,
                        [](auto row) { return row % 5; },
                        [](auto row) { return row % 7; },
                        [](auto row) { return row % 5; },
                        null_every(9))
                });
            testRowResize(row_vector, bits::kNotNull);

            // Resize on lazy children will result in an exception.
            auto rowWithLazyChild = make_row_vector({
                make_lazy_flat_vector<int32_t>(
                    10,
                    [&](vector_size_t i) { return i % 5; },
                    [](vector_size_t i) { return i % 7 == 0; })
            });
            EXPECT_THROW(rowWithLazyChild->resize(20), PolluxException);
        }

        TEST_F(VectorTest, wrapConstantInDictionary) {
            // Wrap Constant in Dictionary with no extra nulls. Expect Constant.
            auto indices = make_indices(10, [](auto row) { return row % 2; });
            auto vector = BaseVector::wrap_in_dictionary(
                nullptr,
                indices,
                10,
                BaseVector::create_constant(INTEGER(), 7, 100, pool()));
            ASSERT_EQ(vector->encoding(), VectorEncoding::Simple::CONSTANT);
            auto constant_vector =
                    std::dynamic_pointer_cast<ConstantVector<int32_t> >(vector);
            for (auto i = 0; i < 10; ++i) {
                ASSERT_FALSE(constant_vector->is_null_at(i));
                ASSERT_EQ(7, constant_vector->value_at(i));
            }

            // Wrap Constant in Dictionary with extra nulls. Expect Dictionary.
            auto nulls = make_nulls(10, [](auto row) { return row % 3 == 0; });
            vector = BaseVector::wrap_in_dictionary(
                nulls,
                indices,
                10,
                BaseVector::create_constant(INTEGER(), 11, 100, pool()));
            ASSERT_EQ(vector->encoding(), VectorEncoding::Simple::DICTIONARY);
            auto dictVector = std::dynamic_pointer_cast<SimpleVector<int32_t> >(vector);
            for (auto i = 0; i < 10; ++i) {
                if (i % 3 == 0) {
                    ASSERT_TRUE(dictVector->is_null_at(i));
                } else {
                    ASSERT_FALSE(dictVector->is_null_at(i));
                    ASSERT_EQ(11, dictVector->value_at(i));
                }
            }
        }

        TEST_F(VectorTest, setFlatVectorStringView) {
            auto vector = BaseVector::create(VARCHAR(), 1, pool());
            auto flat = vector->as_flat_vector<StringView>();
            EXPECT_EQ(0, flat->stringBuffers().size());

            const std::string originalString = "This string is too long to be inlined";

            flat->set(0, StringView(originalString));
            EXPECT_EQ(flat->value_at(0).getString(), originalString);
            EXPECT_EQ(1, flat->stringBuffers().size());
            EXPECT_EQ(originalString.size(), flat->stringBuffers()[0]->size());

            // Make a copy of the vector. Verify that string buffer is shared.
            auto copy = BaseVector::create(VARCHAR(), 1, pool());
            copy->copy(flat, 0, 0, 1);

            auto flatCopy = copy->as_flat_vector<StringView>();
            EXPECT_EQ(1, flatCopy->stringBuffers().size());
            EXPECT_EQ(flat->stringBuffers()[0].get(), flatCopy->stringBuffers()[0].get());

            // Modify the string in the copy. Make sure it is written into a new string
            // buffer.
            const std::string newString = "A different string";
            flatCopy->set(0, StringView(newString));
            EXPECT_EQ(2, flatCopy->stringBuffers().size());
            EXPECT_EQ(flat->stringBuffers()[0].get(), flatCopy->stringBuffers()[0].get());
            EXPECT_EQ(originalString.size(), flatCopy->stringBuffers()[0]->size());
            EXPECT_EQ(newString.size(), flatCopy->stringBuffers()[1]->size());
        }

        TEST_F(VectorTest, resizeAtConstruction) {
            const size_t realSize = 10;

            vector_size_t oldByteSize = BaseVector::byteSize<int64_t>(realSize);
            BufferPtr values = AlignedBuffer::allocate<char>(oldByteSize, pool());

            EXPECT_EQ(oldByteSize, values->size());
            EXPECT_GE(values->capacity(), values->size());

            // Resize the vector.
            const size_t newByteSize = oldByteSize / 2;
            const size_t curCapacity = values->capacity();
            values->setSize(newByteSize);

            // Check it has been resized.
            EXPECT_EQ(newByteSize, values->size());
            EXPECT_EQ(curCapacity, values->capacity());

            // Now create a FlatVector with the resized buffer.
            auto flat = std::make_shared<FlatVector<int64_t> >(
                pool(),
                BIGINT(),
                BufferPtr(nullptr),
                realSize,
                std::move(values),
                std::vector<BufferPtr>());

            // Check that the vector was resized back to the old size.
            EXPECT_EQ(oldByteSize, flat->values()->size());
        }

        TEST_F(VectorTest, resizeStringAsciiness) {
            std::vector<std::string> stringInput = {"hellow", "how", "are"};
            auto flat_vector = make_flat_vector(stringInput);
            auto stringVector = flat_vector->as<SimpleVector<StringView> >();
            SelectivityVector rows(stringInput.size());
            stringVector->computeAndSetIsAscii(rows);
            ASSERT_TRUE(stringVector->isAscii(rows).value());
            stringVector->resize(2);
            ASSERT_FALSE(stringVector->isAscii(rows));
        }

        TEST_F(VectorTest, resizeZeroString) {
            auto vector = make_flat_vector<std::string>(
                {
                    "This is a string",
                    "This is another string",
                    "This is the third string"
                });
            ASSERT_EQ(1, vector->stringBuffers().size());
            ASSERT_LT(0, vector->stringBuffers()[0]->size());

            const auto capacity = vector->stringBuffers()[0]->capacity();
            ASSERT_GT(capacity, 0);

            vector->resize(0);
            ASSERT_EQ(1, vector->stringBuffers().size());
            ASSERT_EQ(0, vector->stringBuffers()[0]->size());
            ASSERT_EQ(capacity, vector->stringBuffers()[0]->capacity());
        }

        TEST_F(VectorTest, copyNoRows) { {
                auto source = make_flat_vector<int32_t>({1, 2, 3});
                auto target = BaseVector::create(INTEGER(), 10, pool());
                SelectivityVector rows(3, false);
                target->copy(source.get(), rows, nullptr);
            } {
                auto source = make_flat_vector<StringView>({"a", "b", "c"});
                auto target = BaseVector::create(VARCHAR(), 10, pool());
                SelectivityVector rows(3, false);
                target->copy(source.get(), rows, nullptr);
            }
        }

        TEST_F(VectorTest, copyAscii) {
            std::vector<std::string> stringData = {"a", "b", "c"};
            auto source = make_flat_vector(stringData);
            SelectivityVector all(stringData.size());
            source->setAllIsAscii(true);

            auto other = make_flat_vector(stringData);
            other->copy(source.get(), all, nullptr);
            auto ascii = other->isAscii(all);
            ASSERT_TRUE(ascii.has_value());
            ASSERT_TRUE(ascii.value());

            // Copy over asciness from a vector without asciiness set.
            source->invalidateIsAscii();
            other->copy(source.get(), all, nullptr);
            // Ensure isAscii returns nullopt
            ascii = other->isAscii(all);
            ASSERT_FALSE(ascii.has_value());

            // Now copy over some rows from vector without ascii compute on it.
            other->setAllIsAscii(false);
            SelectivityVector some(all.size(), false);
            some.setValid(1, true);
            some.updateBounds();
            other->copy(source.get(), some, nullptr);
            // We will have invalidated all.
            ascii = other->isAscii(all);
            ASSERT_FALSE(ascii.has_value());

            auto largerSource = make_flat_vector<std::string>({"a", "b", "c", "d"});
            vector_size_t sourceMappings[] = {3, 2, 1, 0};
            some.setAll();
            some.setValid(0, false);
            some.updateBounds();

            other->setAllIsAscii(false);
            largerSource->setAllIsAscii(true);
            other->copy(largerSource.get(), some, sourceMappings);
            ascii = other->isAscii(all);
            ASSERT_TRUE(ascii.has_value());
            ASSERT_FALSE(ascii.value());
        }

        TEST_F(VectorTest, compareNan) {
            // Double input.
            {
                std::vector<double> doubleInput = {1.1, std::nan("nan"), 0.9};

                auto flatVector1 = make_flat_vector(doubleInput);
                auto flatVector2 = make_flat_vector(doubleInput);

                for (size_t i = 0; i < doubleInput.size(); ++i) {
                    EXPECT_TRUE(flatVector1->equal_value_at(flatVector2.get(), i, i));
                }
            }

            // Float input.
            {
                std::vector<float> floatInput = {1.1, std::nanf("nan"), 0.9};

                auto flatVector1 = make_flat_vector(floatInput);
                auto flatVector2 = make_flat_vector(floatInput);

                for (size_t i = 0; i < floatInput.size(); ++i) {
                    EXPECT_TRUE(flatVector1->equal_value_at(flatVector2.get(), i, i));
                }
            }
        }

        class VectorCreateConstantTest : public VectorTest {
        public:
            template<TypeKind KIND>
            void testPrimitiveConstant(
                typename TypeTraits<KIND>::NativeType val,
                const TypePtr &type = Type::create<KIND>()) {
                using TCpp = typename TypeTraits<KIND>::NativeType;
                variant var = variant::create<KIND>(val);

                auto baseVector = BaseVector::create_constant(type, var, size_, pool());
                auto simpleVector = baseVector->template as<SimpleVector<TCpp> >();
                ASSERT_TRUE(simpleVector != nullptr);

                ASSERT_EQ(KIND, simpleVector->type_kind());
                ASSERT_EQ(size_, simpleVector->size());
                ASSERT_TRUE(simpleVector->is_scalar());

                for (auto i = 0; i < simpleVector->size(); i++) {
                    if constexpr (std::is_same_v<TCpp, StringView>) {
                        ASSERT_EQ(
                            var.template value<KIND>(), std::string(simpleVector->value_at(i)));
                    } else {
                        ASSERT_EQ(var.template value<KIND>(), simpleVector->value_at(i));
                    }
                }

                verifyConstantToString(type, baseVector);
            }

            template<TypeKind KIND>
            void testComplexConstant(const TypePtr &type, const VectorPtr &vector) {
                ASSERT_EQ(KIND, type->kind());
                auto baseVector = BaseVector::wrap_in_constant(size_, 0, vector);
                auto simpleVector = baseVector->template as<
                    SimpleVector<typename KindToFlatVector<KIND>::WrapperType> >();
                ASSERT_TRUE(simpleVector != nullptr);

                ASSERT_EQ(KIND, simpleVector->type_kind());
                ASSERT_EQ(size_, simpleVector->size());
                ASSERT_FALSE(simpleVector->is_scalar());

                for (auto i = 0; i < simpleVector->size(); i++) {
                    ASSERT_TRUE(vector->equal_value_at(baseVector.get(), 0, i));
                }

                verifyConstantToString(type, baseVector);
            }

            void verifyConstantToString(const TypePtr &type, const VectorPtr &constant) {
                auto expectedStr = fmt::format(
                    "[CONSTANT {}: {} elements, {}]",
                    type->toString(),
                    size_,
                    constant->toString(0));
                EXPECT_EQ(expectedStr, constant->toString());
                for (auto i = 1; i < constant->size(); ++i) {
                    EXPECT_EQ(constant->toString(0), constant->toString(i));
                }
            }

            template<TypeKind KIND>
            void testNullConstant(const TypePtr &type) {
                ASSERT_EQ(KIND, type->kind());
                auto baseVector = BaseVector::create_null_constant(type, size_, pool());
                auto simpleVector = baseVector->template as<
                    SimpleVector<typename KindToFlatVector<KIND>::WrapperType> >();
                ASSERT_TRUE(simpleVector != nullptr);

                ASSERT_EQ(KIND, simpleVector->type_kind());
                ASSERT_TRUE(type->equivalent(*simpleVector->type()));
                ASSERT_EQ(size_, simpleVector->size());
                ASSERT_EQ(simpleVector->is_scalar(), TypeTraits<KIND>::isPrimitiveType);

                for (auto i = 0; i < simpleVector->size(); i++) {
                    ASSERT_TRUE(simpleVector->is_null_at(i));
                }

                auto expectedStr = fmt::format(
                    "[CONSTANT {}: {} elements, null]", type->toString(), size_);
                EXPECT_EQ(expectedStr, baseVector->toString());
                for (auto i = 1; i < baseVector->size(); ++i) {
                    EXPECT_EQ(baseVector->toString(0), baseVector->toString(i));
                }
            }

        protected:
            // Theoretical "size" of the constant vector created.
            const size_t size_{23};
        };

        TEST_F(VectorCreateConstantTest, scalar) {
            testPrimitiveConstant<TypeKind::BIGINT>(-123456789);
            testPrimitiveConstant<TypeKind::INTEGER>(98765);
            testPrimitiveConstant<TypeKind::SMALLINT>(1234);
            testPrimitiveConstant<TypeKind::TINYINT>(123);

            testPrimitiveConstant<TypeKind::BOOLEAN>(true);
            testPrimitiveConstant<TypeKind::BOOLEAN>(false);

            testPrimitiveConstant<TypeKind::REAL>(99.98);
            testPrimitiveConstant<TypeKind::DOUBLE>(12.345);

            testPrimitiveConstant<TypeKind::VARCHAR>(StringView("hello world"));
            testPrimitiveConstant<TypeKind::VARBINARY>(StringView("my binary buffer"));
        }

        TEST_F(VectorCreateConstantTest, complex) {
            testComplexConstant<TypeKind::ARRAY>(
                ARRAY(INTEGER()),
                make_array_vector<int32_t>(
                    1, [](auto) { return 10; }, [](auto i) { return i; }));

            testComplexConstant<TypeKind::MAP>(
                MAP(INTEGER(), REAL()),
                make_map_vector<int32_t, float>(
                    1,
                    [](auto) { return 10; },
                    [](auto i) { return i; },
                    [](auto i) { return i; }));

            testComplexConstant<TypeKind::ROW>(
                ROW({{"c0", INTEGER()}}),
                make_row_vector({make_flat_vector<int32_t>(1, [](auto i) { return i; })}));
        }

        TEST_F(VectorCreateConstantTest, null) {
            testNullConstant<TypeKind::BIGINT>(BIGINT());
            testNullConstant<TypeKind::INTEGER>(INTEGER());
            testNullConstant<TypeKind::SMALLINT>(SMALLINT());
            testNullConstant<TypeKind::TINYINT>(TINYINT());

            testNullConstant<TypeKind::BOOLEAN>(BOOLEAN());

            testNullConstant<TypeKind::REAL>(REAL());
            testNullConstant<TypeKind::DOUBLE>(DOUBLE());
            testNullConstant<TypeKind::BIGINT>(DECIMAL(10, 5));
            testNullConstant<TypeKind::HUGEINT>(DECIMAL(20, 5));

            testNullConstant<TypeKind::TIMESTAMP>(TIMESTAMP());
            testNullConstant<TypeKind::INTEGER>(DATE());
            testNullConstant<TypeKind::BIGINT>(INTERVAL_DAY_TIME());

            testNullConstant<TypeKind::VARCHAR>(VARCHAR());
            testNullConstant<TypeKind::VARBINARY>(VARBINARY());

            testNullConstant<TypeKind::ROW>(ROW({BIGINT(), REAL()}));
            testNullConstant<TypeKind::ARRAY>(ARRAY(DOUBLE()));
            testNullConstant<TypeKind::MAP>(MAP(INTEGER(), DOUBLE()));
        }

        class TestingHook : public ValueHook {
        public:
            TestingHook(RowSet rows, SimpleVector<int64_t> *values)
                : rows_(rows), values_(values) {
            }

            bool acceptsNulls() const override {
                return false;
            }

            void addValue(vector_size_t row, int64_t value) override {
                if (values_->is_null_at(rows_[row])) {
                    ++errors_;
                } else {
                    auto original = values_->value_at(rows_[row]);
                    if (original != value) {
                        ++errors_;
                    }
                }
            }

            int32_t errors() const {
                return errors_;
            }

        private:
            RowSet rows_;
            SimpleVector<int64_t> *values_;
            int32_t errors_ = 0;
        };

        TEST_F(VectorTest, valueHook) {
            VectorPtr values = createScalar<TypeKind::BIGINT>(BIGINT(), 1000, true);
            auto lazy = std::make_shared<LazyVector>(
                pool(),
                BIGINT(),
                values->size(),
                std::make_unique<TestingLoader>(values));
            std::vector<vector_size_t> rows;
            for (int i = 0; i < values->size(); i += 2) {
                rows.push_back(i);
            }
            TestingHook hook(rows, values->as<SimpleVector<int64_t> >());
            // The hook is called with the row number being an index 'i' into
            // 'rows' and the value being the value at rows[i]. The hook is not
            // called if the value at the row is null because the hook does not
            // accept nulls.  In this way, for the use case of aggregation, the
            // row number is directly usable as an index into an array of groups
            // to update. The caller will compose the RowSet based on what rows
            // have been filtered out beteen creating the LazyVector and the
            // point of loading. The hook itself is concerned with the loaded
            // subset of rows and the position of each row within this set.
            lazy->load(rows, &hook);
            EXPECT_EQ(hook.errors(), 0);
        }

        TEST_F(VectorTest, byteSize) {
            constexpr vector_size_t count = std::numeric_limits<vector_size_t>::max();
            constexpr uint64_t expected = count * sizeof(int64_t);
            EXPECT_EQ(BaseVector::byteSize<int64_t>(count), expected);
        }

        TEST_F(VectorTest, clear_nulls) {
            auto vectorSize = 100;
            auto vector = BaseVector::create(INTEGER(), vectorSize, pool());
            ASSERT_FALSE(vector->may_have_nulls());

            // No op if doesn't have nulls
            SelectivityVector selection{vectorSize};
            vector->clear_nulls(selection);
            ASSERT_FALSE(vector->may_have_nulls());

            // De-allocate nulls if all selected
            auto rawNulls = vector->mutable_raw_nulls();
            ASSERT_EQ(bits::count_nulls(rawNulls, 0, vectorSize), 0);
            bits::set_null(rawNulls, 50);
            ASSERT_TRUE(vector->is_null_at(50));
            ASSERT_EQ(bits::count_nulls(rawNulls, 0, vectorSize), 1);
            vector->clear_nulls(selection);
            ASSERT_FALSE(vector->may_have_nulls());

            // Clear within vectorSize
            rawNulls = vector->mutable_raw_nulls();
            bits::set_null(rawNulls, 50);
            bits::set_null(rawNulls, 70);
            selection.clearAll();
            selection.setValidRange(40, 60, true);
            selection.updateBounds();
            vector->clear_nulls(selection);
            ASSERT_TRUE(!vector->is_null_at(50));
            ASSERT_TRUE(vector->is_null_at(70));

            // Clear with end > vector size
            selection.resize(120);
            selection.clearAll();
            selection.setValidRange(60, 120, true);
            selection.updateBounds();
            vector->clear_nulls(selection);
            ASSERT_TRUE(!vector->is_null_at(70));
            ASSERT_TRUE(vector->may_have_nulls());

            // Clear with begin > vector size
            rawNulls = vector->mutable_raw_nulls();
            bits::set_null(rawNulls, 70);
            selection.clearAll();
            selection.setValidRange(100, 120, true);
            selection.updateBounds();
            vector->clear_nulls(selection);
            ASSERT_TRUE(vector->is_null_at(70));
        }

        TEST_F(VectorTest, setStringToNull) {
            constexpr int32_t kSize = 100;
            auto target = make_flat_vector<StringView>(
                kSize, [](auto /*row*/) { return StringView("Non-inlined string"); });
            target->set_null(kSize - 1, true);
            auto unknownNull = std::make_shared<ConstantVector<UnknownValue> >(
                pool(), kSize, true, UNKNOWN(), UnknownValue());

            auto stringNull = BaseVector::wrap_in_constant(kSize, kSize - 1, target);
            SelectivityVector rows(kSize, false);
            rows.setValid(2, true);
            rows.updateBounds();
            target->copy(unknownNull.get(), rows, nullptr);
            EXPECT_TRUE(target->is_null_at(2));

            rows.setValid(4, true);
            rows.updateBounds();
            target->copy(stringNull.get(), rows, nullptr);
            EXPECT_TRUE(target->is_null_at(4));
            auto nulls = allocate_nulls(kSize, pool(), bits::kNull);
            auto flatNulls = std::make_shared<FlatVector<UnknownValue> >(
                pool(),
                UNKNOWN(),
                nulls,
                kSize,
                BufferPtr(nullptr),
                std::vector<BufferPtr>());
            rows.setValid(6, true);
            rows.updateBounds();
            target->copy(flatNulls.get(), rows, nullptr);
            EXPECT_TRUE(target->is_null_at(6));
            EXPECT_EQ(4, bits::count_nulls(target->raw_nulls(), 0, kSize));
        }

        TEST_F(VectorTest, clear_all_nulls) {
            auto vectorSize = 100;
            auto vector = BaseVector::create(INTEGER(), vectorSize, pool());
            ASSERT_FALSE(vector->may_have_nulls());

            auto rawNulls = vector->mutable_raw_nulls();
            ASSERT_EQ(bits::count_nulls(rawNulls, 0, vectorSize), 0);
            bits::set_null(rawNulls, 50);
            ASSERT_TRUE(vector->is_null_at(50));
            ASSERT_EQ(bits::count_nulls(rawNulls, 0, vectorSize), 1);
            vector->clear_all_nulls();
            ASSERT_FALSE(vector->may_have_nulls());
            ASSERT_FALSE(vector->is_null_at(50));
        }

        TEST_F(VectorTest, constantSetNull) {
            auto vector = make_constant<int64_t>(123, 10);

            POLLUX_ASSERT_THROW(
                vector->set_null(0, true), "set_null not supported on ConstantVector");
        }

        /// Test lazy vector wrapped in multiple layers of dictionaries.
        TEST_F(VectorTest, multipleDictionariesOverLazy) {
            vector_size_t size = 10;
            auto indices = make_indices(size, [&](auto row) { return size - row - 1; });
            auto lazy = std::make_shared<LazyVector>(
                pool(),
                INTEGER(),
                size,
                std::make_unique<TestingLoader>(
                    make_flat_vector<int32_t>(size, [](auto row) { return row; })));

            auto dict = BaseVector::wrap_in_dictionary(
                nullptr,
                indices,
                size,
                BaseVector::wrap_in_dictionary(nullptr, indices, size, lazy));

            dict->loaded_vector();
            for (auto i = 0; i < size; i++) {
                ASSERT_EQ(i, dict->as<SimpleVector<int32_t>>()->value_at(i));
            }
        }

        /// Test lazy loading of nested dictionary vector
        TEST_F(VectorTest, selectiveLoadingOfLazyDictionaryNested) {
            vector_size_t size = 10;
            auto indices =
                    make_indices(size, [&](auto row) { return (row % 2 == 0) ? row : 0; });
            auto data = make_flat_vector<int32_t>(size, [](auto row) { return row; });

            auto loader = std::make_unique<TestingLoader>(data);
            auto loaderPtr = loader.get();
            auto lazyVector =
                    std::make_shared<LazyVector>(pool(), INTEGER(), size, std::move(loader));

            auto indicesInner =
                    make_indices(size, [&](auto row) { return (row % 2 == 0) ? row : 0; });
            auto indicesOuter =
                    make_indices(size, [&](auto row) { return (row % 4 == 0) ? row : 0; });

            auto dict = BaseVector::wrap_in_dictionary(
                nullptr,
                indicesOuter,
                size,
                BaseVector::wrap_in_dictionary(nullptr, indicesInner, size, lazyVector));

            dict->loaded_vector();
            ASSERT_EQ(loaderPtr->rowCount(), 1 + size / 4);

            dict->loaded_vector();
            ASSERT_EQ(loaderPtr->rowCount(), 1 + size / 4);
        }

        TEST_F(VectorTest, nestedLazy) {
            // Verify that explicit checks are triggered that ensure lazy vectors cannot
            // be nested within two different top level vectors.
            vector_size_t size = 10;
            auto indexAt = [](vector_size_t) { return 0; };
            auto makeLazy = [&]() {
                return std::make_shared<LazyVector>(
                    pool(),
                    INTEGER(),
                    size,
                    std::make_unique<TestingLoader>(
                        make_flat_vector<int64_t>(size, [](auto row) { return row; })));
            };
            auto lazy = makeLazy();
            auto dict = BaseVector::wrap_in_dictionary(
                nullptr, make_indices(size, indexAt), size, lazy);

            POLLUX_ASSERT_THROW(
                BaseVector::wrap_in_dictionary(
                    nullptr, make_indices(size, indexAt), size, lazy),
                "An unloaded lazy vector cannot be wrapped by two different top level"
                " vectors.");

            // Verify that if the original dictionary layer is destroyed without loading
            // the underlying vector then the lazy vector can be wrapped in a new encoding
            // layer.
            dict.reset();
            dict = BaseVector::wrap_in_dictionary(
                nullptr, make_indices(size, indexAt), size, lazy);

            // Verify that the unloaded dictionary can be nested as long as it has one top
            // level vector.
            EXPECT_NO_THROW(BaseVector::wrap_in_dictionary(
                nullptr, make_indices(size, indexAt), size, dict));

            // Limitation: Current checks cannot prevent existing references of the lazy
            // vector to load rows. For example, the following would succeed even though
            // it was wrapped under 2 layer of dictionary above:
            EXPECT_FALSE(lazy->isLoaded());
            EXPECT_NO_THROW(lazy->loaded_vector());
            EXPECT_TRUE(lazy->isLoaded());
        }

        TEST_F(VectorTest, wrapInDictionaryOverLoadedLazy) {
            // Ensure the lazy layer is stripped away and the dictionaryValues vector
            // points to the loaded Vector underneath it.
            vector_size_t size = 10;
            auto lazy = std::make_shared<LazyVector>(
                pool(),
                INTEGER(),
                size,
                std::make_unique<TestingLoader>(
                    make_flat_vector<int64_t>(size, [](auto row) { return row; })));
            lazy->loaded_vector();
            EXPECT_TRUE(lazy->isLoaded());
            auto dict = wrap_in_dictionary(make_indices(size, melon::identity), size, lazy);
            auto valuesVector = dict->value_vector();
            EXPECT_FALSE(valuesVector->is_lazy());
        }

        TEST_F(VectorTest, dictionaryResize) {
            vector_size_t size = 10;
            std::vector<int64_t> elements{0, 1, 2, 3, 4};
            auto makeIndicesFn = [&]() {
                return make_indices(size, [&](auto row) { return row % elements.size(); });
            };

            auto indices = makeIndicesFn();
            auto flat_vector = make_flat_vector<int64_t>(elements);

            // Create a simple dictionary.
            auto dict = wrap_in_dictionary(std::move(indices), size, flat_vector);

            auto expectedValues = std::vector<int64_t>{0, 1, 2, 3, 4, 0, 1, 2, 3, 4};
            auto expectedVector = make_flat_vector<int64_t>(expectedValues);
            test::assertEqualVectors(expectedVector, dict);

            // Double size.
            dict->resize(size * 2);

            // Check all the newly resized indices point to 0th value.
            expectedVector = make_flat_vector<int64_t>(
                {0, 1, 2, 3, 4, 0, 1, 2, 3, 4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0});
            test::assertEqualVectors(expectedVector, dict);

            // Resize  a nested dictionary.
            auto innerDict = wrap_in_dictionary(makeIndicesFn(), size, flat_vector);
            auto outerDict = wrap_in_dictionary(makeIndicesFn(), size, innerDict);

            expectedVector = make_flat_vector<int64_t>(expectedValues);
            test::assertEqualVectors(expectedVector, outerDict);
            innerDict->resize(size * 2);
            // Check that the outer dictionary remains unaffected.
            test::assertEqualVectors(expectedVector, outerDict);

            // Resize a shared nested dictionary with shared indices.
            indices = makeIndicesFn();
            dict = wrap_in_dictionary(
                indices, size, wrap_in_dictionary(indices, size, flat_vector));

            ASSERT_TRUE(!indices->unique());
            dict->resize(size * 2);
            expectedVector = make_flat_vector<int64_t>(
                {0, 1, 2, 3, 4, 0, 1, 2, 3, 4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0});
            test::assertEqualVectors(expectedVector, dict);
            // Check to ensure that indices has not changed.
            ASSERT_EQ(indices->size(), size * sizeof(vector_size_t));

            // Check that resize clear indices even if no new allocation happens.
            {
                auto indicesLarge = make_indices(20, [&](auto row) { return 3; });
                auto *rawIndices = indicesLarge->as<vector_size_t>();
                auto dictVector = wrap_in_dictionary(indicesLarge, 10, flat_vector);

                // Release reference to 'indicesLarge' to make it single-referenced and
                // allow reuse in dictVector->resize().
                indicesLarge.reset();
                dictVector->resize(15);

                for (int i = 0; i < 10; i++) {
                    EXPECT_EQ(rawIndices[i], 3);
                }
                for (int i = 10; i < 15; i++) {
                    EXPECT_EQ(rawIndices[i], 0);
                }
                for (int i = 15; i < 20; i++) {
                    EXPECT_EQ(rawIndices[i], 3);
                }
            }
        }

        TEST_F(VectorTest, acquireSharedStringBuffers) {
            const int numBuffers = 10;
            std::vector<BufferPtr> buffers;
            const int bufferSize = 100;
            for (int i = 0; i < numBuffers; ++i) {
                buffers.push_back(AlignedBuffer::allocate<char>(bufferSize, pool()));
            }
            auto vector = BaseVector::create(VARCHAR(), 100, pool());
            auto flat_vector = vector->as<FlatVector<StringView> >();
            EXPECT_EQ(0, flat_vector->stringBuffers().size());

            flat_vector->setStringBuffers(buffers);
            EXPECT_EQ(numBuffers, flat_vector->stringBuffers().size());
            for (int i = 0; i < numBuffers; ++i) {
                EXPECT_EQ(buffers[i], flat_vector->stringBuffers()[i]);
            }

            flat_vector->setStringBuffers({});
            EXPECT_EQ(0, flat_vector->stringBuffers().size());

            int numSourceVectors = 2;
            std::vector<VectorPtr> sourceVectors;
            for (int i = 0; i < numSourceVectors; ++i) {
                sourceVectors.push_back(BaseVector::create(VARCHAR(), 100, pool()));
                sourceVectors.back()->as_flat_vector<StringView>()->setStringBuffers(
                    {buffers[i]});
            }
            flat_vector->setStringBuffers({buffers[0]});
            EXPECT_EQ(1, flat_vector->stringBuffers().size());
            flat_vector->acquireSharedStringBuffers(sourceVectors[0].get());
            EXPECT_EQ(1, flat_vector->stringBuffers().size());
            flat_vector->acquireSharedStringBuffers(sourceVectors[1].get());
            EXPECT_EQ(2, flat_vector->stringBuffers().size());

            flat_vector->acquireSharedStringBuffers(sourceVectors[0].get());
            flat_vector->acquireSharedStringBuffers(sourceVectors[1].get());
            EXPECT_EQ(2, flat_vector->stringBuffers().size());
            for (int i = 0; i < numSourceVectors; ++i) {
                EXPECT_EQ(buffers[i], flat_vector->stringBuffers()[i]);
            }

            // insert with duplicate buffers and expect an exception.
            flat_vector->setStringBuffers({buffers[0], buffers[0]});
            EXPECT_EQ(2, flat_vector->stringBuffers().size());
            flat_vector->acquireSharedStringBuffers(sourceVectors[0].get());
            EXPECT_EQ(2, flat_vector->stringBuffers().size());

            // Function does not throw if the input is not varchar or varbinary.
            flat_vector->setStringBuffers({buffers[0]});
            auto array_vector = make_array_vector<int32_t>({});
            ASSERT_NO_THROW(flat_vector->acquireSharedStringBuffers(array_vector.get()));

            auto unkownVector = BaseVector::create(UNKNOWN(), 100, pool());
            ASSERT_NO_THROW(flat_vector->acquireSharedStringBuffers(unkownVector.get()));
        }

        TEST_F(VectorTest, acquireSharedStringBuffersRecursive) {
            auto vector = BaseVector::create(VARCHAR(), 100, pool());
            auto flat_vector = vector->as<FlatVector<StringView> >();

            auto testWithEncodings = [&](const VectorPtr &source,
                                         const std::function<void()> &check) {
                flat_vector->setStringBuffers({});
                flat_vector->acquireSharedStringBuffersRecursive(source.get());
                check();

                // Constant vector.
                auto constant_vector = BaseVector::wrap_in_constant(10, 0, source);
                flat_vector->setStringBuffers({});
                flat_vector->acquireSharedStringBuffersRecursive(constant_vector.get());
                check();

                // Dictionary Vector.
                BufferPtr indices =
                        AlignedBuffer::allocate<vector_size_t>(source->size(), pool());
                for (int32_t i = 0; i < source->size(); ++i) {
                    indices->asMutable<vector_size_t>()[i] = source->size() - i - 1;
                }
                auto dictionary_vector =
                        BaseVector::wrap_in_dictionary(nullptr, indices, source->size(), source);
                flat_vector->setStringBuffers({});
                flat_vector->acquireSharedStringBuffersRecursive(dictionary_vector.get());
                check();
            };

            // Acquiring buffer from array

            {
                // Array<int>
                auto array_vector = make_array_vector<int32_t>({{1, 2, 3}});
                testWithEncodings(array_vector, [&]() {
                    ASSERT_EQ(flat_vector->stringBuffers().size(), 0);
                });
            } {
                // Array<Varchar>
                auto array_vector = make_array_vector<StringView>(
                    {{"This is long enough not to be inlined !!!", "b"}});

                testWithEncodings(array_vector, [&]() {
                    EXPECT_EQ(1, flat_vector->stringBuffers().size());
                    ASSERT_EQ(
                        flat_vector->stringBuffers()[0],
                        array_vector->as<ArrayVector>()
                        ->elements()
                        ->as_flat_vector<StringView>()
                        ->stringBuffers()[0]);
                });
            }

            // Array<Array<Varchar>>
            auto array_vector = make_nullable_nested_array_vector<StringView>(
                {{{{{"This is long enough not to be inlined !!!"_sv}}}}});

            testWithEncodings(array_vector, [&]() {
                EXPECT_EQ(1, flat_vector->stringBuffers().size());
                ASSERT_EQ(
                    flat_vector->stringBuffers()[0],
                    array_vector->as<ArrayVector>()
                    ->elements()
                    ->as<ArrayVector>()
                    ->elements()
                    ->as_flat_vector<StringView>()
                    ->stringBuffers()[0]);
            });

            // Map<Varchar,Varchar>
            auto map_vector = make_map_vector<StringView, StringView>(
                {
                    {
                        {
                            "This is long enough not to be inlined !!!"_sv,
                            "This is long enough not to be inlined !!!"_sv
                        }
                    }
                });

            testWithEncodings(map_vector, [&]() {
                EXPECT_EQ(2, flat_vector->stringBuffers().size());
                ASSERT_EQ(
                    flat_vector->stringBuffers()[0],
                    map_vector->as<MapVector>()
                    ->mapKeys()
                    ->as_flat_vector<StringView>()
                    ->stringBuffers()[0]);
                ASSERT_EQ(
                    flat_vector->stringBuffers()[1],
                    map_vector->as<MapVector>()
                    ->mapValues()
                    ->as_flat_vector<StringView>()
                    ->stringBuffers()[0]);
            });

            // Row
            flat_vector->setStringBuffers({});
            auto row_vector =
                    make_row_vector({map_vector, array_vector, make_flat_vector<int32_t>({1, 2})});
            testWithEncodings(
                row_vector, [&]() { EXPECT_EQ(3, flat_vector->stringBuffers().size()); });
        }

        /// Test MapVector::canonicalize for a MapVector with 'values' vector shorter
        /// than 'keys' vector.
        TEST_F(VectorTest, mapCanonicalizeValuesShorterThenKeys) {
            auto map_vector = std::make_shared<MapVector>(
                pool(),
                MAP(BIGINT(), BIGINT()),
                nullptr,
                2,
                make_indices({0, 2}),
                make_indices({2, 2}),
                make_flat_vector<int64_t>({7, 6, 5, 4, 3, 2, 1}),
                make_flat_vector<int64_t>({6, 5, 4, 3, 2, 1}));
            EXPECT_FALSE(map_vector->hasSortedKeys());

            MapVector::canonicalize(map_vector);
            EXPECT_TRUE(map_vector->hasSortedKeys());

            // Verify that keys and values referenced from the map are sorted. Keys and
            // values not referenced from the map are not sorted.
            test::assertEqualVectors(
                make_flat_vector<int64_t>({6, 7, 4, 5, 3, 2}), map_vector->mapKeys());
            test::assertEqualVectors(
                make_flat_vector<int64_t>({5, 6, 3, 4, 2, 1}), map_vector->mapValues());
        }

        TEST_F(VectorTest, mapCanonicalize) {
            auto map_vector = make_map_vector<int64_t, int64_t>({
                {{4, 40}, {3, 30}, {2, 20}, {5, 50}, {1, 10}},
                {{4, 41}},
                {},
            });

            EXPECT_FALSE(map_vector->hasSortedKeys());

            test::assertEqualVectors(
                map_vector->mapKeys(), make_flat_vector<int64_t>({4, 3, 2, 5, 1, 4}));
            test::assertEqualVectors(
                map_vector->mapValues(),
                make_flat_vector<int64_t>({40, 30, 20, 50, 10, 41}));

            // Sort keys.
            MapVector::canonicalize(map_vector);
            EXPECT_TRUE(map_vector->hasSortedKeys());

            test::assertEqualVectors(
                map_vector->mapKeys(), make_flat_vector<int64_t>({1, 2, 3, 4, 5, 4}));
            test::assertEqualVectors(
                map_vector->mapValues(),
                make_flat_vector<int64_t>({10, 20, 30, 40, 50, 41}));

            EXPECT_EQ(map_vector->sizeAt(0), 5);
            EXPECT_EQ(map_vector->offsetAt(0), 0);

            EXPECT_EQ(map_vector->sizeAt(1), 1);
            EXPECT_EQ(map_vector->offsetAt(1), 5);

            EXPECT_EQ(map_vector->sizeAt(2), 0);
            EXPECT_EQ(map_vector->offsetAt(2), 6);
        }

        TEST_F(VectorTest, mapCreateSorted) {
            BufferPtr offsets = make_indices({0, 5, 6});
            BufferPtr sizes = make_indices({5, 1, 0});
            auto map_vector = std::make_shared<MapVector>(
                pool(),
                MAP(BIGINT(), BIGINT()),
                nullptr,
                3,
                offsets,
                sizes,
                make_flat_vector<int64_t>({1, 2, 3, 4, 5, 4}),
                make_flat_vector<int64_t>({10, 20, 30, 40, 50, 41}),
                std::nullopt,
                true);

            EXPECT_TRUE(map_vector->hasSortedKeys());
        }

        TEST_F(VectorTest, flatSliceMutability) {
            auto vec = make_flat_vector<int64_t>(
                10,
                [](vector_size_t i) { return i; },
                [](vector_size_t i) { return i % 3 == 0; });
            auto slice = std::dynamic_pointer_cast<FlatVector<int64_t> >(vec->slice(0, 3));
            ASSERT_TRUE(slice);
            slice->set_null(0, false);
            EXPECT_FALSE(slice->is_null_at(0));
            EXPECT_TRUE(vec->is_null_at(0));
            slice = std::dynamic_pointer_cast<FlatVector<int64_t> >(vec->slice(0, 3));
            slice->mutableRawValues()[2] = -1;
            EXPECT_EQ(slice->value_at(2), -1);
            EXPECT_EQ(vec->value_at(2), 2);
        }

        TEST_F(VectorTest, arraySliceMutability) {
            testArrayOrMapSliceMutability(
                std::dynamic_pointer_cast<ArrayVector>(createArray(vectorSize_, false)));
        }

        TEST_F(VectorTest, mapSliceMutability) {
            testArrayOrMapSliceMutability(
                std::dynamic_pointer_cast<MapVector>(createMap(vectorSize_, false)));
        }

        // Demonstrates incorrect usage of the memory pool. The pool is destroyed
        // while the vector allocated from it is still alive.
        TEST_F(VectorTest, lifetime) {
            ASSERT_DEATH(
                {
                auto childPool = memory::memoryManager()->addLeafPool();
                auto v = BaseVector::create(INTEGER(), 10, childPool.get());

                // BUG: Memory pool needs to stay alive until all memory allocated from
                // it is freed.
                childPool.reset();
                v.reset();
                },
                "");
        }

        TEST_F(VectorTest, ensure_nulls_capacity) {
            // Has to be more than 1 byte's worth of bits.
            {
                size_t size = 100;
                auto vec = make_flat_vector<int64_t>(size, [](vector_size_t i) { return i; });
                auto nulls = vec->mutable_nulls(2);
                ASSERT_GE(nulls->size(), bits::nbytes(size));
            }

            // Mutli-referenced nulls.
            {
                auto vector =
                        make_nullable_flat_vector<int32_t>({std::nullopt, 1, std::nullopt, 2});
                auto nulls = vector->nulls();
                ASSERT_FALSE(vector->nulls()->unique());
                vector->mutable_raw_nulls();
                ASSERT_TRUE(vector->nulls()->unique());
            }
        }

        TEST_F(VectorTest, createVectorWithNullType) {
            std::string kErrorMessage = "Vector creation requires a non-null type.";

            POLLUX_ASSERT_THROW(BaseVector::create(nullptr, 100, pool()), kErrorMessage);
            POLLUX_ASSERT_THROW(
                BaseVector::create_null_constant(nullptr, 100, pool()), kErrorMessage);
            POLLUX_ASSERT_THROW(
                BaseVector::get_or_create_empty(nullptr, nullptr, pool()), kErrorMessage);

            POLLUX_ASSERT_THROW(
                std::make_shared<ConstantVector<int64_t>>(pool(), 100, false, nullptr, 0),
                kErrorMessage);

            std::vector<BufferPtr> stringBuffers;
            POLLUX_ASSERT_THROW(
                std::make_shared<FlatVector<int64_t>>(
                    pool(), nullptr, nullptr, 100, nullptr, std::move(stringBuffers)),
                kErrorMessage);

            std::vector<VectorPtr> children;
            POLLUX_ASSERT_THROW(
                std::make_shared<RowVector>(pool(), nullptr, nullptr, 100, children),
                kErrorMessage);

            POLLUX_ASSERT_THROW(RowVector::createEmpty(nullptr, pool()), kErrorMessage);

            POLLUX_ASSERT_THROW(
                std::make_shared<ArrayVector>(
                    pool(), nullptr, nullptr, 100, nullptr, nullptr, nullptr),
                kErrorMessage);

            POLLUX_ASSERT_THROW(
                std::make_shared<MapVector>(
                    pool(), nullptr, nullptr, 100, nullptr, nullptr, nullptr, nullptr),
                kErrorMessage);
        }

        TEST_F(VectorTest, testCopyWithZeroCount) {
            auto runTest = [&](const VectorPtr &vector) {
                // We pass invalid targetIndex and sourceIndex and expect the
                // function not to throw since count is 0.
                ASSERT_NO_THROW(
                    vector->copy(vector.get(), vector->size() + 1, vector->size() + 1, 0));

                BaseVector::CopyRange range{vector->size() + 1, vector->size() + 1, 0};
                ASSERT_NO_THROW(vector->copy_ranges(vector.get(), melon::Range(&range, 1)));

                ASSERT_NO_THROW(vector->copy_ranges(
                    vector.get(), std::vector<BaseVector::CopyRange>{range, range, range}));
            };

            // Flat.
            runTest(make_flat_vector<bool>({1, 0, 1, 1}));
            runTest(make_flat_vector<StringView>({"s"_sv}));
            runTest(make_flat_vector<int32_t>({1, 2}));

            // Complex types.
            runTest(make_array_vector<int32_t>(
                1, [](auto) { return 10; }, [](auto i) { return i; }));

            runTest(make_map_vector<int32_t, float>(
                1,
                [](auto) { return 10; },
                [](auto i) { return i; },
                [](auto i) { return i; }));

            runTest(
                make_row_vector({make_flat_vector<int32_t>(1, [](auto i) { return i; })}));
        }

        TEST_F(VectorTest, flattenVector) {
            auto test = [&](VectorPtr &vector, bool stayTheSame) {
                auto original = vector;
                BaseVector::flattenVector(vector);
                if (stayTheSame) {
                    EXPECT_EQ(original.get(), vector.get());
                } else {
                    EXPECT_NE(original.get(), vector.get());
                }
                test::assertEqualVectors(original, vector);
            };

            // Flat input.
            VectorPtr flat = make_flat_vector<int32_t>({1, 2, 3});
            test(flat, true);

            VectorPtr array = make_array_vector<int32_t>({{1, 2, 3}, {1}, {1}});
            test(array, true);

            VectorPtr map =
                    make_map_vector<int32_t, int32_t>({{{4, 40}, {3, 30}}, {{4, 41}}, {}});
            test(map, true);

            VectorPtr row = make_row_vector({flat, array, map});
            test(row, true);

            VectorPtr lazy = std::make_shared<LazyVector>(
                pool(), INTEGER(), flat->size(), std::make_unique<TestingLoader>(flat));
            test(lazy, true);

            // Constant
            VectorPtr constant = BaseVector::wrap_in_constant(100, 1, flat);
            test(constant, false);
            EXPECT_TRUE(constant->is_flat_encoding());

            // Dictionary
            VectorPtr dictionary = BaseVector::wrap_in_dictionary(
                nullptr, make_indices(100, [](auto row) { return row % 2; }), 100, flat);
            test(dictionary, false);
            EXPECT_TRUE(dictionary->is_flat_encoding());

            VectorPtr lazyDictionary =
                    wrap_in_lazy_dictionary(make_flat_vector<int32_t>({1, 2, 3}));
            test(lazyDictionary, true);
            EXPECT_TRUE(lazyDictionary->is_lazy());
            EXPECT_TRUE(lazyDictionary->loaded_vector()->is_flat_encoding());

            // Array with constant elements.
            auto *array_vector = array->as<ArrayVector>();
            array_vector->elements() = BaseVector::wrap_in_constant(100, 1, flat);
            auto originalElements = array_vector->elements();
            auto original = array;

            BaseVector::flattenVector(array);
            test::assertEqualVectors(original, array);

            EXPECT_EQ(array->encoding(), VectorEncoding::Simple::ARRAY);
            EXPECT_TRUE(array->as<ArrayVector>()->elements()->is_flat_encoding());

            EXPECT_EQ(original.get(), array.get());
            EXPECT_NE(originalElements.get(), array->as<ArrayVector>()->elements().get());

            // Map with constant keys and values.
            auto *map_vector = map->as<MapVector>();
            auto originalValues = map_vector->mapValues() =
                                  BaseVector::wrap_in_constant(100, 1, flat);
            auto originalKeys = map_vector->mapKeys() =
                                BaseVector::wrap_in_constant(100, 1, flat);

            original = map;
            BaseVector::flattenVector(map);
            test::assertEqualVectors(original, map);

            EXPECT_EQ(map->encoding(), VectorEncoding::Simple::MAP);
            EXPECT_TRUE(map->as<MapVector>()->mapValues()->is_flat_encoding());
            EXPECT_TRUE(map->as<MapVector>()->mapKeys()->is_flat_encoding());

            EXPECT_EQ(original.get(), map.get());
            EXPECT_NE(originalValues.get(), map->as<MapVector>()->mapValues().get());
            EXPECT_NE(originalKeys.get(), map->as<MapVector>()->mapKeys().get());

            // Row with constant field.
            row = make_row_vector({flat, BaseVector::wrap_in_constant(3, 1, flat)});
            auto *row_vector = row->as<RowVector>();
            auto originalRow0 = row_vector->children()[0];
            auto originalRow1 = row_vector->children()[1];
            original = row;
            BaseVector::flattenVector(row);
            EXPECT_EQ(row->encoding(), VectorEncoding::Simple::ROW);
            EXPECT_TRUE(row->as<RowVector>()->children()[0]->is_flat_encoding());
            EXPECT_TRUE(row->as<RowVector>()->children()[1]->is_flat_encoding());
            test::assertEqualVectors(original, row);

            EXPECT_EQ(original.get(), row.get());
            EXPECT_EQ(originalRow0.get(), row->as<RowVector>()->children()[0].get());
            EXPECT_NE(originalRow1.get(), row->as<RowVector>()->children()[1].get());

            // Row with constant field.
            // Null input
            VectorPtr nullVector = nullptr;
            BaseVector::flattenVector(nullVector);
            EXPECT_EQ(nullVector, nullptr);
        }

        TEST_F(VectorTest, find_duplicate_value) {
            const CompareFlags flags;
            auto data = make_flat_vector<int64_t>({1, 3, 2, 4, 3, 5, 4, 6});

            // No duplicates in the first 4 values.
            auto dup = data->find_duplicate_value(0, 4, flags);
            ASSERT_FALSE(dup.has_value());

            // No duplicates in the last 4 values.
            dup = data->find_duplicate_value(4, 4, flags);
            ASSERT_FALSE(dup.has_value());

            // No duplicates in rows 2 to 6.
            dup = data->find_duplicate_value(2, 4, flags);
            ASSERT_FALSE(dup.has_value());

            // Values in rows 1 and 4 are the same.
            dup = data->find_duplicate_value(0, 5, flags);
            ASSERT_TRUE(dup.has_value());
            ASSERT_EQ(4, dup.value());

            // Values in rows 3 and 7 are the same.
            dup = data->find_duplicate_value(2, 6, flags);
            ASSERT_TRUE(dup.has_value());
            ASSERT_EQ(6, dup.value());

            // Verify that empty range doesn't throw.
            dup = data->find_duplicate_value(2, 0, flags);
            ASSERT_FALSE(dup.has_value());

            // Verify that out-of-bound range throws in debug mode.
#ifndef NDEBUG
            POLLUX_ASSERT_THROW(
                data->find_duplicate_value(2, 8, flags), "Size is too large");
            POLLUX_ASSERT_THROW(
                data->find_duplicate_value(22, 8, flags), "Start index is too large");
#endif

            // Verify that empty vector doesn't throw.
            auto noData = make_flat_vector<int32_t>({});
            dup = noData->find_duplicate_value(0, 0, flags);
            ASSERT_FALSE(dup.has_value());
        }

        TEST_F(VectorTest, resizeArrayAndMapResetOffsets) {
            auto checkIndices = [&](const vector_size_t *indices) {
                ASSERT_EQ(indices[0], 1);
                ASSERT_EQ(indices[1], 1);
                ASSERT_EQ(indices[2], 0);
                ASSERT_EQ(indices[3], 0);
            };

            // Test array.
            {
                auto offsets = make_indices({0, 1, 2, 3});
                auto sizes = make_indices({1, 1, 1, 1});

                auto *rawSizes = sizes->as<vector_size_t>();

                auto array_vector = std::make_shared<ArrayVector>(
                    pool(),
                    ARRAY(BIGINT()),
                    nullptr,
                    2,
                    offsets,
                    sizes,
                    make_flat_vector<int64_t>({1, 2, 3, 4}));

                // Release references to 'offsets' and 'sizes' to make them
                // single-referenced and allow reuse in array_vector->resize().
                offsets.reset();
                sizes.reset();

                array_vector->resize(4);
                checkIndices(rawSizes);
            }

            // Test map.
            {
                auto offsets = make_indices({0, 1, 2, 3});
                auto sizes = make_indices({1, 1, 1, 1});

                auto *rawSizes = sizes->as<vector_size_t>();

                auto map_vector = std::make_shared<MapVector>(
                    pool(),
                    MAP(BIGINT(), BIGINT()),
                    nullptr,
                    2,
                    offsets,
                    sizes,
                    make_flat_vector<int64_t>({1, 2, 3, 4}),
                    make_flat_vector<int64_t>({1, 2, 3, 4}));

                // Release references to 'offsets' and 'sizes' to make them
                // single-referenced and allow reuse in map_vector->resize().
                offsets.reset();
                sizes.reset();

                map_vector->resize(4);
                checkIndices(rawSizes);
            }
        }

        TEST_F(VectorTest, getRawStringBufferWithSpace) {
            auto vector =
                    make_flat_vector<StringView>({"ee", "rr", "rryy", "12345678901234"});
            ASSERT_EQ(vector->stringBuffers().size(), 1);
            auto originalBufferSize = vector->stringBuffers().back()->size();
            std::string replace = "I'm replace 123456789";
            char *rawBuffer = vector->getRawStringBufferWithSpace(replace.size());
            ASSERT_EQ(vector->stringBuffers().size(), 1);
            memcpy(rawBuffer, replace.data(), replace.size());
            vector->setNoCopy(1, StringView(rawBuffer, replace.size()));

            auto lastBuffer = vector->stringBuffers().back();
            ASSERT_EQ(originalBufferSize + replace.size(), lastBuffer->size());
            auto expected = make_flat_vector<StringView>(
                {"ee", "I'm replace 123456789", "rryy", "12345678901234"});
            test::assertEqualVectors(expected, vector);

            // Use up all but 5 bytes in the 'lastBuffer', then ask for buffer with at
            // least 6 bytes of space. Expect a fairly large new buffer.
            lastBuffer->setSize(lastBuffer->capacity() - 5);
            rawBuffer = vector->getRawStringBufferWithSpace(6);
            ASSERT_EQ(vector->stringBuffers().size(), 2);

            lastBuffer = vector->stringBuffers().back();
            ASSERT_EQ(6, lastBuffer->size());
            ASSERT_EQ(49056, lastBuffer->capacity());

            // Use up all bytes in 'lastBuffer, then ask for buffer with exactly one byte
            // of space. Expect a small new buffer.
            lastBuffer->setSize(lastBuffer->capacity());

            rawBuffer = vector->getRawStringBufferWithSpace(1, true /*exactSize*/);
            ASSERT_EQ(vector->stringBuffers().size(), 3);
            lastBuffer = vector->stringBuffers().back();
            ASSERT_EQ(1, lastBuffer->size());
            ASSERT_EQ(32, lastBuffer->capacity());
        }

        TEST_F(VectorTest, getRawStringBufferWithSpaceNoExistingBuffer) {
            auto vector = make_flat_vector<StringView>({"ee", "rr", "rryy", "kk"});
            ASSERT_EQ(vector->stringBuffers().size(), 0);
            std::string replace = "I'm replace 123456789";
            char *rawBuffer = vector->getRawStringBufferWithSpace(replace.size());
            ASSERT_EQ(vector->stringBuffers().size(), 1);
            memcpy(rawBuffer, replace.data(), replace.size());
            vector->setNoCopy(1, StringView(rawBuffer, replace.size()));

            auto lastBuffer = vector->stringBuffers().back();
            ASSERT_EQ(replace.size(), lastBuffer->size());
            auto expected =
                    make_flat_vector<StringView>({"ee", "I'm replace 123456789", "rryy", "kk"});
            test::assertEqualVectors(expected, vector);
        }

        namespace {
            SelectivityVector toSelectivityVector(
                const std::vector<BaseVector::CopyRange> &ranges) {
                vector_size_t size = 0;
                for (const auto &range: ranges) {
                    POLLUX_CHECK_EQ(range.sourceIndex, range.targetIndex);
                    if (range.sourceIndex >= size) {
                        size = range.sourceIndex + range.count;
                    }
                }

                SelectivityVector rows(size, false);
                for (const auto &range: ranges) {
                    rows.setValidRange(
                        range.sourceIndex, range.sourceIndex + range.count, true);
                }
                rows.updateBounds();
                return rows;
            }
        } // namespace

        TEST_F(VectorTest, mutableIndices) {
            // Ensure that DictionaryVector::mutableIndices always returns a new indices
            // buffer if its not unique.
            auto flat_vector = make_flat_vector<int64_t>({1, 2, 3, 4, 5, 6, 7, 8});
            auto indices = make_indices(8, [](auto row) { return row % 2; });
            auto vector = BaseVector::wrap_in_dictionary(nullptr, indices, 8, flat_vector);
            ASSERT_EQ(vector->encoding(), VectorEncoding::Simple::DICTIONARY);
            auto dictionary = vector->as<DictionaryVector<int64_t> >();

            auto mutableIndices = dictionary->mutableIndices(1);
            ASSERT_NE(indices.get(), mutableIndices.get());
            auto mutableIndicesPtr = mutableIndices.get();
            // Reset the ptr so we can get the same indices when calling mutableIndices.
            mutableIndices.reset();
            auto mutableIndicesAgain = dictionary->mutableIndices(1);
            ASSERT_EQ(mutableIndicesPtr, mutableIndicesAgain.get());

            // Ensure that mutable indices are different.
            mutableIndices = dictionary->mutableIndices(1);
            ASSERT_NE(mutableIndices.get(), mutableIndicesAgain.get());
        }

        TEST_F(VectorTest, to_copy_ranges) {
            SelectivityVector rows(10);

            auto testRoundTrip = [&]() {
                auto ranges = BaseVector::to_copy_ranges(rows);
                ASSERT_EQ(rows, toSelectivityVector(ranges));
            };

            // All selected.
            testRoundTrip();

            // One row missing in the middle.
            rows.setValid(3, false);
            rows.updateBounds();

            testRoundTrip();

            // One row at each end is missing.
            rows.setValid(0, false);
            rows.setValid(9, false);
            rows.updateBounds();

            testRoundTrip();

            // Nothing selected.
            rows.clearAll();
            testRoundTrip();

            // Only every 3-rd row selected.
            for (auto i = 0; i < 10; i += 3) {
                rows.setValid(i, true);
            }
            rows.updateBounds();
            testRoundTrip();

            // Every 3-rd row is not selected.
            rows.setAll();
            for (auto i = 0; i < 10; i += 3) {
                rows.setValid(i, false);
            }
            rows.updateBounds();
            testRoundTrip();
        }

        TEST_F(VectorTest, rowCopyRanges) {
            RowVectorPtr rowVectorDest = make_row_vector(
                {make_flat_vector<int32_t>({1, 2}), make_flat_vector<int32_t>({1, 2})});
            RowVectorPtr RowVectorSrc = make_row_vector(
                {make_flat_vector<int32_t>({3, 4}), make_flat_vector<int32_t>({3, 4})});
            std::vector<BaseVector::CopyRange> baseRanges{
                {
                    .sourceIndex = 0,
                    .targetIndex = 2,
                    .count = 2,
                }
            };
            rowVectorDest->resize(4);

            // Make sure nulls overwritten.
            rowVectorDest->set_null(2, true);
            rowVectorDest->set_null(3, true);

            rowVectorDest->copy_ranges(RowVectorSrc.get(), baseRanges);
            auto expected = make_row_vector(
                {
                    make_flat_vector<int32_t>({1, 2, 3, 4}),
                    make_flat_vector<int32_t>({1, 2, 3, 4})
                });
            test::assertEqualVectors(expected, rowVectorDest);
        }

        TEST_F(VectorTest, rowCopyRangesWithGap) {
            // Hold an extra reference to c0 so it becomes immutable.
            auto targetC0 = make_flat_vector<int64_t>({1, 2, 3});
            auto target = make_row_vector({targetC0});
            auto source = make_row_vector({make_flat_vector<int64_t>({4, 5})});
            BaseVector::CopyRange ranges[2];
            ranges[0].targetIndex = 0;
            ranges[0].sourceIndex = 1;
            ranges[0].count = 1;
            ranges[1].targetIndex = 2;
            ranges[1].sourceIndex = 0;
            ranges[1].count = 1;
            target->copy_ranges(
                source.get(), melon::Range(std::begin(ranges), std::end(ranges)));
            auto expected = make_row_vector({make_flat_vector<int64_t>({5, 2, 4})});
            test::assertEqualVectors(expected, target);
        }

        TEST_F(VectorTest, containsNullAtIntegers) {
            VectorPtr data = make_flat_vector<int32_t>({1, 2, 3});
            for (auto i = 0; i < data->size(); ++i) {
                EXPECT_FALSE(data->contains_null_at(i));
            }

            auto indices = make_indices({0, 0, 1, 1, 2, 2});
            auto dictionary = wrap_in_dictionary(indices, data);

            for (auto i = 0; i < dictionary->size(); ++i) {
                EXPECT_FALSE(dictionary->contains_null_at(i));
            }

            auto nulls = make_nulls({true, false, true, false, true, false});
            dictionary = BaseVector::wrap_in_dictionary(nulls, indices, 6, data);
            for (auto i = 0; i < dictionary->size(); i += 2) {
                EXPECT_TRUE(dictionary->contains_null_at(i));
            }

            for (auto i = 1; i < dictionary->size(); i += 2) {
                EXPECT_FALSE(dictionary->contains_null_at(i));
            }

            data = make_nullable_flat_vector<int32_t>({1, std::nullopt, 3});
            EXPECT_FALSE(data->contains_null_at(0));
            EXPECT_TRUE(data->contains_null_at(1));
            EXPECT_FALSE(data->contains_null_at(2));

            dictionary = wrap_in_dictionary(indices, data);
            EXPECT_FALSE(dictionary->contains_null_at(0));
            EXPECT_FALSE(dictionary->contains_null_at(1));
            EXPECT_TRUE(dictionary->contains_null_at(2));
            EXPECT_TRUE(dictionary->contains_null_at(3));
            EXPECT_FALSE(dictionary->contains_null_at(4));
            EXPECT_FALSE(dictionary->contains_null_at(5));

            dictionary = BaseVector::wrap_in_dictionary(nulls, indices, 6, data);
            EXPECT_TRUE(dictionary->contains_null_at(0));
            EXPECT_FALSE(dictionary->contains_null_at(1));
            EXPECT_TRUE(dictionary->contains_null_at(2));
            EXPECT_TRUE(dictionary->contains_null_at(3));
            EXPECT_TRUE(dictionary->contains_null_at(4));
            EXPECT_FALSE(dictionary->contains_null_at(5));

            data = make_constant(10, 3);
            for (auto i = 0; i < data->size(); ++i) {
                EXPECT_FALSE(data->contains_null_at(i));
            }

            data = make_null_constant(TypeKind::INTEGER, 3);
            for (auto i = 0; i < data->size(); ++i) {
                EXPECT_TRUE(data->contains_null_at(i));
            }
        }

        TEST_F(VectorTest, containsNullAtArrays) {
            auto data = make_nullable_array_vector<int32_t>({
                {{1, 2}},
                {{1, 2, std::nullopt, 3}},
                common::testutil::optionalEmpty,
                std::nullopt,
                {{1, 2, 3, 4}},
            });

            EXPECT_FALSE(data->contains_null_at(0));
            EXPECT_TRUE(data->contains_null_at(1));
            EXPECT_FALSE(data->contains_null_at(2));
            EXPECT_TRUE(data->contains_null_at(3));
            EXPECT_FALSE(data->contains_null_at(4));
        }

        TEST_F(VectorTest, containsNullAtMaps) {
            auto data = make_nullable_map_vector<int32_t, int64_t>({
                {{{1, 10}, {2, 20}}},
                {{{3, 30}}},
                {{{1, 10}, {2, 20}, {3, std::nullopt}, {4, 40}}},
                common::testutil::optionalEmpty,
                std::nullopt,
                {{{1, 10}, {2, 20}, {3, 30}, {4, 40}}},
            });

            EXPECT_FALSE(data->contains_null_at(0));
            EXPECT_FALSE(data->contains_null_at(1));
            EXPECT_TRUE(data->contains_null_at(2));
            EXPECT_FALSE(data->contains_null_at(3));
            EXPECT_TRUE(data->contains_null_at(4));
            EXPECT_FALSE(data->contains_null_at(5));
        }

        TEST_F(VectorTest, containsNullAtStructs) {
            auto data = make_row_vector(
                {
                    make_nullable_flat_vector<int32_t>({1, 2, std::nullopt, 4, 5}),
                    make_nullable_flat_vector<int64_t>(
                        {10, std::nullopt, std::nullopt, 40, 50}),
                    make_nullable_flat_vector<int16_t>({1, 2, 3, 4, 5}),
                    make_nullable_flat_vector<int16_t>({10, 20, 30, 40, 50}),
                },
                [](auto row) { return row == 3; });

            EXPECT_FALSE(data->contains_null_at(0));
            EXPECT_TRUE(data->contains_null_at(1));
            EXPECT_TRUE(data->contains_null_at(2));
            EXPECT_TRUE(data->contains_null_at(3));
            EXPECT_FALSE(data->contains_null_at(4));

            data = make_row_vector(
                {
                    make_nullable_flat_vector<int32_t>({1, 2, 3, 4, 5, 6}),
                    make_nullable_array_vector<int64_t>({
                        {{1, 2}},
                        {{1, 2, std::nullopt, 3}},
                        common::testutil::optionalEmpty,
                        {{1, 2, 3}},
                        std::nullopt,
                        {{1, 2, 3, 4, 5}},
                    }),
                },
                [](auto row) { return row == 3; });

            EXPECT_FALSE(data->contains_null_at(0));
            EXPECT_TRUE(data->contains_null_at(1));
            EXPECT_FALSE(data->contains_null_at(2));
            EXPECT_TRUE(data->contains_null_at(3));
            EXPECT_TRUE(data->contains_null_at(4));
            EXPECT_FALSE(data->contains_null_at(5));
        }

        TEST_F(VectorTest, mutableValues) {
            auto vector = make_flat_vector<int64_t>(1'000, [](auto row) { return row; });

            auto *rawValues = vector->rawValues();
            vector->mutableValues(1'001);
            ASSERT_EQ(rawValues, vector->rawValues());
            for (auto i = 0; i < 1'000; ++i) {
                EXPECT_EQ(rawValues[i], i);
            }

            vector->mutableValues(10'000);
            ASSERT_NE(rawValues, vector->rawValues());
            rawValues = vector->rawValues();
            for (auto i = 0; i < 1'000; ++i) {
                EXPECT_EQ(rawValues[i], i);
            }

            auto values = vector->mutableValues(2'000);
            ASSERT_EQ(rawValues, vector->rawValues());
            for (auto i = 0; i < 1'000; ++i) {
                EXPECT_EQ(rawValues[i], i);
            }

            vector->mutableValues(500);
            ASSERT_NE(rawValues, vector->rawValues());
            rawValues = vector->rawValues();
            for (auto i = 0; i < 500; ++i) {
                EXPECT_EQ(rawValues[i], i);
            }
        }

        TEST_F(VectorTest, appendNulls) {
            auto row_vector = make_row_vector({make_flat_vector<int32_t>({0, 1, 2})});
            ASSERT_EQ(row_vector->size(), 3);

            // Append 0 rows.
            row_vector->appendNulls(0);
            ASSERT_EQ(row_vector->size(), 3);
            test::assertEqualVectors(
                row_vector, make_row_vector({make_flat_vector<int32_t>({0, 1, 2})}));

            // Append 2 nulls
            row_vector->appendNulls(2);
            ASSERT_EQ(row_vector->size(), 5);
            ASSERT_TRUE(row_vector->is_null_at(3));
            ASSERT_TRUE(row_vector->is_null_at(4));
            test::assertEqualVectors(
                row_vector->childAt(0), make_flat_vector<int32_t>({0, 1, 2}));

            // Append negative.
            EXPECT_ANY_THROW(row_vector->appendNulls(-1));
        }

        TEST_F(VectorTest, primitiveTypeNullEqual) {
            auto base = make_nullable_flat_vector<int32_t>({1, 2, std::nullopt});
            auto other = make_nullable_flat_vector<int32_t>({1, std::nullopt, 1});

            auto equalNoStop = [&](vector_size_t i, vector_size_t j) {
                return base
                        ->equal_value_at(
                            other.get(), i, j, CompareFlags::NullHandlingMode::kNullAsValue)
                        .value();
            };

            auto equalStopAtNull = [&](vector_size_t i, vector_size_t j) {
                return base->equal_value_at(
                    other.get(),
                    i,
                    j,
                    CompareFlags::NullHandlingMode::kNullAsIndeterminate);
            };

            // No null compare.
            ASSERT_TRUE(equalNoStop(0, 0));
            ASSERT_TRUE(equalStopAtNull(0, 0).value());

            // Null compare in NullAsValue mode.
            ASSERT_FALSE(equalNoStop(1, 1));
            ASSERT_FALSE(equalNoStop(2, 2));

            // Null compare in StopAtNull mode.
            ASSERT_FALSE(equalStopAtNull(1, 1).has_value());
            ASSERT_FALSE(equalStopAtNull(2, 2).has_value());
        }

        TEST_F(VectorTest, complexTypeNullEqual) {
            auto base =
                    make_array_vector_from_json<int32_t>({"[0, 1]", "[2, 2]", "[2, null]"});
            auto other =
                    make_array_vector_from_json<int32_t>({"[0, 1]", "[2, null]", "[1, 2]"});
            auto equalNoStop = [&](vector_size_t i, vector_size_t j) {
                return base
                        ->equal_value_at(
                            other.get(), i, j, CompareFlags::NullHandlingMode::kNullAsValue)
                        .value();
            };

            auto equalStopAtNull = [&](vector_size_t i, vector_size_t j) {
                return base->equal_value_at(
                    other.get(),
                    i,
                    j,
                    CompareFlags::NullHandlingMode::kNullAsIndeterminate);
            };

            // No null compare, [0, 1] vs [0, 1].
            ASSERT_TRUE(equalNoStop(0, 0));
            ASSERT_TRUE(equalStopAtNull(0, 0).value());

            // No null compare, [2, null] vs [1, 2].
            ASSERT_FALSE(equalNoStop(2, 2));
            ASSERT_FALSE(equalStopAtNull(2, 2).value());

            // Null compare in NullAsValue mode, [2, 2] vs [2, null].
            ASSERT_FALSE(equalNoStop(1, 1));

            // Null compare in StopAtNull mode, [2, 2] vs [2, null].
            ASSERT_FALSE(equalStopAtNull(1, 1).has_value());
        }

        TEST_F(VectorTest, dictionaryNullEqual) {
            auto base =
                    make_array_vector_from_json<int32_t>({"[0, 1]", "[2, 2]", "[2, null]"});
            auto other =
                    make_array_vector_from_json<int32_t>({"[0, 1]", "[2, null]", "[1, 2]"});
            auto baseVectorSize = base->size();
            auto kTopLevelVectorSize = baseVectorSize * 2;
            BufferPtr indices = kumo::pollux::make_indices(
                kTopLevelVectorSize,
                [&](vector_size_t i) { return i % baseVectorSize; },
                pool());
            auto dictVector =
                    BaseVector::wrap_in_dictionary(nullptr, indices, kTopLevelVectorSize, base);

            auto equalNoStop = [&](vector_size_t i, vector_size_t j) {
                return dictVector
                        ->equal_value_at(
                            other.get(), i, j, CompareFlags::NullHandlingMode::kNullAsValue)
                        .value();
            };

            auto equalStopAtNull = [&](vector_size_t i, vector_size_t j) {
                return dictVector->equal_value_at(
                    other.get(),
                    i,
                    j,
                    CompareFlags::NullHandlingMode::kNullAsIndeterminate);
            };

            for (vector_size_t i = 0; i < 2; ++i) {
                // No null compare, [0, 1] vs [0, 1].
                ASSERT_TRUE(equalNoStop(i * baseVectorSize, 0));
                ASSERT_TRUE(equalStopAtNull(i * baseVectorSize, 0).value());

                // No null compare, [2, null] vs [1. 2].
                ASSERT_FALSE(equalNoStop(2 + i * baseVectorSize, 2));
                ASSERT_FALSE(equalStopAtNull(2 + i * baseVectorSize, 2).value());

                // Null compare in NullAsValue mode, [2, 2] vs [2, null].
                ASSERT_FALSE(equalNoStop(1 + i * baseVectorSize, 1));

                // Null compare in StopAtNull mode, [2, 2] vs [2, null].
                ASSERT_FALSE(equalStopAtNull(1 + i * baseVectorSize, 1).has_value());
            }
        }

        TEST_F(VectorTest, constantNullEqual) {
            auto base =
                    make_array_vector_from_json<int32_t>({"[0, 1]", "[2, 2]", "[2, null]"});
            auto other =
                    make_array_vector_from_json<int32_t>({"[0, 1]", "[2, null]", "[1, 2]"});
            auto baseVectorSize = base->size();
            auto kTopLevelVectorSize = baseVectorSize * 2;
            // [2, null]
            auto constant_vector =
                    BaseVector::wrap_in_constant(kTopLevelVectorSize, 2, base);

            auto equalNoStop = [&](vector_size_t i, vector_size_t j) {
                return constant_vector
                        ->equal_value_at(
                            other.get(), i, j, CompareFlags::NullHandlingMode::kNullAsValue)
                        .value();
            };

            auto equalStopAtNull = [&](vector_size_t i, vector_size_t j) {
                return constant_vector->equal_value_at(
                    other.get(),
                    i,
                    j,
                    CompareFlags::NullHandlingMode::kNullAsIndeterminate);
            };

            // No null compare, [2, null] vs [0, 1], [2, null] vs [1, 2].
            ASSERT_FALSE(equalNoStop(0, 0));
            ASSERT_FALSE(equalStopAtNull(0, 2).value());

            // Null compare in NullAsValue mode, [2, null] vs [2, null].
            ASSERT_TRUE(equalNoStop(0, 1));

            // Null compare in StopAtNull mode, [2, null] vs [2, null].
            ASSERT_FALSE(equalStopAtNull(0, 1).has_value());
        }

        TEST_F(VectorTest, dictionaryLoadedVectorRemoveLazy) {
            auto makeTestVector = [&]() {
                auto lazy = make_lazy_flat_vector<int64_t>(3, [&](auto row) { return row; });
                auto dictVector =
                        BaseVector::wrap_in_dictionary(nullptr, make_indices({0, 1, 2}), 3, lazy);
                return dictVector;
            }; {
                auto vector = makeTestVector();
                vector->loaded_vector();
                EXPECT_FALSE(
                    vector->as<DictionaryVector<int64_t>>()->value_vector()->is_lazy());
            } {
                auto vector = makeTestVector();
                LazyVector::ensureLoadedRows(vector, SelectivityVector(3));
                EXPECT_FALSE(
                    vector->as<DictionaryVector<int64_t>>()->value_vector()->is_lazy());
            } {
                auto vector = make_row_vector({makeTestVector()});
                vector->loaded_vector();
                EXPECT_FALSE(vector->as<RowVector>()
                    ->childAt(0)
                    ->as<DictionaryVector<int64_t>>()
                    ->value_vector()
                    ->is_lazy());
            } {
                auto vector = make_row_vector({makeTestVector()});
                LazyVector::ensureLoadedRows(vector, SelectivityVector(3));
                EXPECT_FALSE(vector->as<RowVector>()
                    ->childAt(0)
                    ->as<DictionaryVector<int64_t>>()
                    ->value_vector()
                    ->is_lazy());
            }
        }

        template<typename T>
        FlatVectorPtr<T> makeFlatNullValues(
            vector_size_t size,
            const TypePtr &type,
            BufferPtr nulls,
            memory::MemoryPool *pool) {
            return std::make_shared<FlatVector<T> >(
                pool,
                type,
                nulls,
                size,
                nullptr, // values
                std::vector<BufferPtr>{});
        }

        TEST_F(VectorTest, flatAllNulls) {
            // Create FlatVector of all nulls with null values buffer.
            vector_size_t size = 1'000;

            POLLUX_ASSERT_THROW(
                makeFlatNullValues<int64_t>(
                    size, BIGINT(), allocate_nulls(size, pool()), pool()),
                "FlatVector with null values buffer must have all rows set to null")

            auto nulls = allocate_nulls(size, pool(), bits::kNull);

            // BIGINT. set API.
            {
                auto flat = makeFlatNullValues<int64_t>(size, BIGINT(), nulls, pool());

                for (auto i = 0; i < size; ++i) {
                    ASSERT_TRUE(flat->is_null_at(i));
                }

                // Change some rows to non-null.
                flat->set(7, 123LL);
                ASSERT_FALSE(flat->is_null_at(7));
                ASSERT_EQ(123LL, flat->value_at(7));

                for (auto i = 0; i < size; ++i) {
                    if (i != 7) {
                        ASSERT_TRUE(flat->is_null_at(i));
                    }
                }
            }

            // BIGINT. set_null API.
            {
                auto flat = makeFlatNullValues<int64_t>(size, BIGINT(), nulls, pool());

                for (auto i = 0; i < size; ++i) {
                    ASSERT_TRUE(flat->is_null_at(i));
                }

                // Change some rows to non-null.
                flat->set_null(7, false);
                ASSERT_FALSE(flat->is_null_at(7));
                ASSERT_NO_THROW(flat->value_at(7));

                for (auto i = 0; i < size; ++i) {
                    if (i != 7) {
                        ASSERT_TRUE(flat->is_null_at(i));
                    }
                }
            }

            // BOOLEAN.
            {
                auto flat = makeFlatNullValues<bool>(size, BIGINT(), nulls, pool());

                for (auto i = 0; i < size; ++i) {
                    ASSERT_TRUE(flat->is_null_at(i));
                }

                // Change some rows to non-null.
                flat->set(7, true);
                ASSERT_FALSE(flat->is_null_at(7));
                ASSERT_EQ(true, flat->value_at(7));

                flat->set(11, false);
                ASSERT_FALSE(flat->is_null_at(11));
                ASSERT_EQ(false, flat->value_at(11));

                for (auto i = 0; i < size; ++i) {
                    if (i != 7 && i != 11) {
                        ASSERT_TRUE(flat->is_null_at(i));
                    }
                }
            }

            // VARCHAR with set.
            {
                auto flat = makeFlatNullValues<StringView>(size, VARCHAR(), nulls, pool());

                for (auto i = 0; i < size; ++i) {
                    ASSERT_TRUE(flat->is_null_at(i));
                }

                // Change some rows to non-null.
                flat->set(7, "Testing is time consuming");
                ASSERT_FALSE(flat->is_null_at(7));
                ASSERT_EQ("Testing is time consuming", flat->value_at(7).str());

                for (auto i = 0; i < size; ++i) {
                    if (i != 7) {
                        ASSERT_TRUE(flat->is_null_at(i));
                    }
                }
            }

            // VARCHAR with setNoCopy.
            {
                auto flat = makeFlatNullValues<StringView>(size, VARCHAR(), nulls, pool());

                for (auto i = 0; i < size; ++i) {
                    ASSERT_TRUE(flat->is_null_at(i));
                }

                // Change some rows to non-null.
                flat->setNoCopy(7, "Short");
                ASSERT_FALSE(flat->is_null_at(7));
                ASSERT_EQ("Short", flat->value_at(7).str());

                for (auto i = 0; i < size; ++i) {
                    if (i != 7) {
                        ASSERT_TRUE(flat->is_null_at(i));
                    }
                }
            }
        }

        TEST_F(VectorTest, hash_all) {
            auto data = make_flat_vector<int32_t>({1, 2, 3});
            ASSERT_TRUE(data->get_null_count().has_value());

            auto hashes = data->hash_all();

            // Make a similar vector, but without stats, e.g. nullCount unset.
            auto copy = std::make_shared<FlatVector<int32_t> >(
                pool(), INTEGER(), nullptr, 3, data->values(), std::vector<BufferPtr>{});
            ASSERT_FALSE(copy->get_null_count().has_value());

            auto hashesCopy = copy->hash_all();

            for (auto i = 0; i < 3; ++i) {
                ASSERT_EQ(hashes->value_at(i), hashesCopy->value_at(i));
            }
        }

        TEST_F(VectorTest, set_type) {
            auto test = [&](auto &type, auto &newType, auto &invalidNewType) {
                auto vector = BaseVector::create(type, 1'000, pool());

                vector->set_type(newType);
                EXPECT_EQ(vector->type()->toString(), newType->toString());

                POLLUX_ASSERT_RUNTIME_THROW(
                    vector->set_type(invalidNewType),
                    fmt::format(
                        "Cannot change vector type from {} to {}. The old and new types can be different logical types, but the underlying physical types must match."
                        ,
                        newType->toString(),
                        invalidNewType->toString()));
            };

            // ROW
            auto type = ROW({"aa"}, {BIGINT()});
            auto newType = ROW({"bb"}, {BIGINT()});
            auto invalidNewType = ROW({"bb"}, {VARCHAR()});
            test(type, newType, invalidNewType);

            // ROW(ROW)
            type = ROW({"a", "b"}, {ROW({"c", "d"}, {BIGINT(), BIGINT()}), BIGINT()});
            newType =
                    ROW({"a", "b"}, {ROW({"cc", "dd"}, {BIGINT(), BIGINT()}), BIGINT()});
            invalidNewType =
                    ROW({"a", "b"}, {ROW({"cc", "dd"}, {VARCHAR(), BIGINT()}), BIGINT()});
            test(type, newType, invalidNewType);

            // ARRAY(ROW)
            type =
                    ROW({"a", "b"}, {ARRAY(ROW({"c", "d"}, {BIGINT(), BIGINT()})), BIGINT()});
            newType = ROW(
                {"a", "b"}, {ARRAY(ROW({"cc", "dd"}, {BIGINT(), BIGINT()})), BIGINT()});
            invalidNewType = ROW(
                {"a", "b"}, {ARRAY(ROW({"cc", "dd"}, {VARCHAR(), BIGINT()})), BIGINT()});
            test(type, newType, invalidNewType);

            // MAP(ROW)
            type =
                    ROW({"a", "b"},
                        {
                            MAP(ROW({"c", "d"}, {BIGINT(), BIGINT()}),
                                ROW({"e", "f"}, {BIGINT(), BIGINT()})),
                            BIGINT()
                        });
            newType =
                    ROW({"a", "b"},
                        {
                            MAP(ROW({"cc", "dd"}, {BIGINT(), BIGINT()}),
                                ROW({"ee", "ff"}, {BIGINT(), BIGINT()})),
                            BIGINT()
                        });
            invalidNewType =
                    ROW({"a", "b"},
                        {
                            MAP(ROW({"cc", "dd"}, {VARCHAR(), BIGINT()}),
                                ROW({"ee", "ff"}, {VARCHAR(), BIGINT()})),
                            BIGINT()
                        });
            test(type, newType, invalidNewType);
        }

        TEST_F(VectorTest, getLargeStringBuffer) {
            auto vector = make_flat_vector<StringView>({});
            size_t size = size_t(std::numeric_limits<int32_t>::max()) + 1;
            auto *buffer = vector->getBufferWithSpace(size);
            EXPECT_GE(buffer->capacity(), size);
        }

        TEST_F(VectorTest, mapUpdate) {
            auto base = make_nullable_map_vector<int64_t, int64_t>({
                {{{1, 1}, {2, 1}}},
                common::testutil::optionalEmpty,
                {{{3, 1}}},
                std::nullopt,
                {{{4, 1}}},
            });
            auto update = make_nullable_map_vector<int64_t, int64_t>({
                {{{2, 2}, {3, 2}}},
                {{{4, 2}}},
                common::testutil::optionalEmpty,
                {{{5, 2}}},
                std::nullopt,
            });
            auto expected = make_nullable_map_vector<int64_t, int64_t>({
                {{{2, 2}, {3, 2}, {1, 1}}},
                {{{4, 2}}},
                {{{3, 1}}},
                std::nullopt,
                std::nullopt,
            });
            auto actual = base->update({update});
            ASSERT_EQ(actual->size(), expected->size());
            for (int i = 0; i < actual->size(); ++i) {
                ASSERT_TRUE(actual->equal_value_at(expected.get(), i, i));
            }
        }

        TEST_F(VectorTest, mapUpdateRowKeyType) {
            auto base = make_map_vector(
                {0, 2},
                make_row_vector({
                    make_flat_vector<int64_t>({1, 2}),
                    make_flat_vector<int64_t>({1, 2}),
                }),
                make_flat_vector<int64_t>({1, 1}));
            auto update = make_map_vector(
                {0, 2},
                make_row_vector({
                    make_flat_vector<int64_t>({2, 3}),
                    make_flat_vector<int64_t>({2, 3}),
                }),
                make_flat_vector<int64_t>({2, 2}));
            auto expected = make_map_vector(
                {0, 3},
                make_row_vector({
                    make_flat_vector<int64_t>({1, 2, 3}),
                    make_flat_vector<int64_t>({1, 2, 3}),
                }),
                make_flat_vector<int64_t>({1, 2, 2}));
            auto actual = base->update({update});
            ASSERT_EQ(actual->size(), expected->size());
            for (int i = 0; i < actual->size(); ++i) {
                ASSERT_TRUE(actual->equal_value_at(expected.get(), i, i));
            }
        }

        TEST_F(VectorTest, mapUpdateNullMapValue) {
            auto base = make_nullable_map_vector<int64_t, int64_t>({
                {{{1, 1}, {2, 1}}},
            });
            auto update = make_nullable_map_vector<int64_t, int64_t>({
                {{{2, std::nullopt}, {3, 2}}},
            });
            auto expected = make_nullable_map_vector<int64_t, int64_t>({
                {{{1, 1}, {2, std::nullopt}, {3, 2}}},
            });
            auto actual = base->update({update});
            ASSERT_EQ(actual->size(), expected->size());
            for (int i = 0; i < actual->size(); ++i) {
                ASSERT_TRUE(actual->equal_value_at(expected.get(), i, i));
            }
        }

        TEST_F(VectorTest, mapUpdateMultipleUpdates) {
            auto base = make_nullable_map_vector<int64_t, int64_t>({
                {{{1, 1}, {2, 1}}},
                common::testutil::optionalEmpty,
                {{{3, 1}}},
                std::nullopt,
                {{{4, 1}}},
            });
            std::vector<MapVectorPtr> updates = {
                make_nullable_map_vector<int64_t, int64_t>({
                    {{{2, 2}, {3, 2}}},
                    {{{4, 2}}},
                    common::testutil::optionalEmpty,
                    {{{5, 2}}},
                    std::nullopt,
                }),
                make_nullable_map_vector<int64_t, int64_t>({
                    {{{3, 3}, {4, 3}}},
                    std::nullopt,
                    common::testutil::optionalEmpty,
                    common::testutil::optionalEmpty,
                    common::testutil::optionalEmpty,
                }),
            };
            auto expected = make_nullable_map_vector<int64_t, int64_t>({
                {{{1, 1}, {2, 2}, {3, 3}, {4, 3}}},
                std::nullopt,
                {{{3, 1}}},
                std::nullopt,
                std::nullopt,
            });
            auto actual = base->update(updates);
            ASSERT_EQ(actual->size(), expected->size());
            for (int i = 0; i < actual->size(); ++i) {
                ASSERT_TRUE(actual->equal_value_at(expected.get(), i, i));
            }
        }

        TEST_F(VectorTest, mapUpdateConstant) {
            auto base = make_nullable_map_vector<int64_t, int64_t>({
                {{{1, 1}, {2, 1}}},
                {{}},
                {{{3, 1}}},
                std::nullopt,
                {{{4, 1}}},
            });
            auto update = BaseVector::wrap_in_constant(
                base->size(), 0, make_map_vector<int64_t, int64_t>({{{2, 2}}}));
            DecodedVector decoded(*update);
            auto actual = base->update(melon::Range(&decoded, 1));
            auto expected = make_nullable_map_vector<int64_t, int64_t>({
                {{{2, 2}, {1, 1}}},
                {{{2, 2}}},
                {{{2, 2}, {3, 1}}},
                std::nullopt,
                {{{2, 2}, {4, 1}}},
            });
            test::assertEqualVectors(expected, actual);
        }

        TEST_F(VectorTest, mapUpdateDictionary) {
            auto base = make_nullable_map_vector<int64_t, int64_t>({
                {{{1, 1}, {2, 1}}},
                {{}},
                {{{3, 1}}},
                std::nullopt,
                {{{4, 1}}},
            });
            auto update = wrap_in_dictionary(
                make_indices({0, 0, 1, 1, 0}),
                make_map_vector<int64_t, int64_t>({{{2, 2}}, {}}));
            DecodedVector decoded(*update);
            auto actual = base->update(melon::Range(&decoded, 1));
            auto expected = make_nullable_map_vector<int64_t, int64_t>({
                {{{2, 2}, {1, 1}}},
                {{{2, 2}}},
                {{{3, 1}}},
                std::nullopt,
                {{{2, 2}, {4, 1}}},
            });
            test::assertEqualVectors(expected, actual);
        }

        TEST_F(VectorTest, pushDictionaryToRowVectorLeaves) {
            auto iota = make_flat_vector<int64_t>(10, melon::identity);
            auto output = RowVector::pushDictionaryToRowVectorLeaves(iota);
            ASSERT_EQ(output, iota);

            auto input = wrap_in_dictionary(make_indices_in_reverse(10), iota);
            output = RowVector::pushDictionaryToRowVectorLeaves(input);
            ASSERT_EQ(output, input); {
                SCOPED_TRACE("General");
                input = wrap_in_dictionary(
                    make_indices_in_reverse(10),
                    make_row_vector({
                        // c0
                        iota,
                        // c1
                        make_row_vector({iota}),
                        // c2
                        BaseVector::wrap_in_dictionary(
                            make_nulls(10, null_every(3)),
                            make_indices_in_reverse(10),
                            10,
                            iota),
                        // c3
                        iota,
                    }));
                output = RowVector::pushDictionaryToRowVectorLeaves(input);
                test::assertEqualVectors(input, output);
                ASSERT_EQ(output->encoding(), VectorEncoding::Simple::ROW);
                auto *outputRow = output->as_unchecked<RowVector>();
                auto &c0 = outputRow->childAt(0);
                ASSERT_EQ(c0->encoding(), VectorEncoding::Simple::DICTIONARY);
                ASSERT_EQ(c0->wrap_info().get(), input->wrap_info().get());
                auto &c1 = outputRow->childAt(1);
                ASSERT_EQ(c1->encoding(), VectorEncoding::Simple::ROW);
                auto *c1Row = c1->as_unchecked<RowVector>();
                auto &c1c0 = c1Row->childAt(0);
                ASSERT_EQ(c1c0->encoding(), VectorEncoding::Simple::DICTIONARY);
                ASSERT_EQ(c1c0->wrap_info().get(), c0->wrap_info().get());
                auto &c2 = outputRow->childAt(2);
                ASSERT_EQ(c2->encoding(), VectorEncoding::Simple::DICTIONARY);
                auto &c3 = outputRow->childAt(3);
                ASSERT_EQ(c3->encoding(), VectorEncoding::Simple::DICTIONARY);
                ASSERT_EQ(c3->wrap_info().get(), c0->wrap_info().get());
            } {
                SCOPED_TRACE("Nested ROW");
                input = wrap_in_dictionary(
                    make_indices_in_reverse(10),
                    make_row_vector({
                        wrap_in_dictionary(
                            make_indices_in_reverse(10),
                            make_row_vector({
                                iota,
                                wrap_in_dictionary(make_indices_in_reverse(10), iota),
                                iota,
                            })),
                    }));
                output = RowVector::pushDictionaryToRowVectorLeaves(input);
                test::assertEqualVectors(input, output);
                auto *c0 =
                        output->as_checked<RowVector>()->childAt(0)->as_checked<RowVector>();
                auto &c0c0 = c0->childAt(0);
                ASSERT_EQ(c0c0->encoding(), VectorEncoding::Simple::DICTIONARY);
                auto &c0c1 = c0->childAt(1);
                ASSERT_EQ(c0c1->encoding(), VectorEncoding::Simple::DICTIONARY);
                auto &c0c2 = c0->childAt(2);
                ASSERT_EQ(c0c2->encoding(), VectorEncoding::Simple::DICTIONARY);
                ASSERT_EQ(c0c0->wrap_info().get(), c0c2->wrap_info().get());
            } {
                SCOPED_TRACE("Nulls over ROW");
                input = wrap_in_dictionary(
                    make_indices_in_reverse(10),
                    make_row_vector({
                        BaseVector::wrap_in_dictionary(
                            make_nulls(10, null_every(7)),
                            make_indices_in_reverse(10),
                            10,
                            make_row_vector({iota, iota})),
                    }));
                output = RowVector::pushDictionaryToRowVectorLeaves(input);
                test::assertEqualVectors(input, output);
                auto *c0 =
                        output->as_checked<RowVector>()->childAt(0)->as_checked<RowVector>();
                auto &c0c0 = c0->childAt(0);
                ASSERT_EQ(c0c0->encoding(), VectorEncoding::Simple::DICTIONARY);
                auto &c0c1 = c0->childAt(1);
                ASSERT_EQ(c0c1->encoding(), VectorEncoding::Simple::DICTIONARY);
            }
        }

        TEST_F(VectorTest, arrayCopyTargetNullOffsets) {
            auto target = BaseVector::create(ARRAY(BIGINT()), 11, pool());
            auto offsetsRef = target->as_unchecked<ArrayVector>()->offsets();
            ASSERT_TRUE(offsetsRef);
            BaseVector::prepare_for_reuse(target, target->size());
            ASSERT_TRUE(target->as_unchecked<ArrayVector>()->offsets());
            auto source = make_array_vector<int64_t>(
                11, [](auto) { return 1; }, [](auto i, auto) { return i; });
            target->copy(source.get(), 0, 0, source->size());
            test::assertEqualVectors(source, target);
        }

        TEST_F(VectorTest, testOverSizedArray) {
            // Verify that flattening an array/map cannot result in a values vector
            // greater than vector_size_t
            auto flat = make_flat_vector<int32_t>(1000, [](auto /*row*/) { return 1; });
            std::vector<vector_size_t> offsets(1, 0);
            auto array = make_array_vector(offsets, flat);
            auto constArray = BaseVector::wrap_in_constant(21474830, 0, array);
            EXPECT_THROW(BaseVector::flattenVector(constArray), PolluxUserError);
        }

        TEST_F(VectorTest, testFlatteningOfRedundantDictionary) {
            // Verify that BaseVector::wrap_in_dictionary() API flattens the output if
            // 'flattenIfRedundant' is set to true and the wrapped vector has a size less
            // than 1/8 the size of the base.
            auto vectorSize = 100;
            auto flat =
                    make_flat_vector<int32_t>(vectorSize, [](auto /*row*/) { return 1; }); {
                auto dictionarySize = vectorSize / 10;
                auto indices = make_indices(dictionarySize, [](auto i) { return i; });
                auto wrapped =
                        BaseVector::wrap_in_dictionary(nullptr, indices, dictionarySize, flat);
                EXPECT_EQ(wrapped->encoding(), VectorEncoding::Simple::DICTIONARY);

                wrapped = BaseVector::wrap_in_dictionary(
                    nullptr, indices, dictionarySize, flat, true /*flattenIfRedundant*/);
                EXPECT_EQ(wrapped->encoding(), VectorEncoding::Simple::FLAT);
            } {
                auto dictionarySize = vectorSize / 3;
                auto indices = make_indices(dictionarySize, [](auto i) { return i; });
                auto wrapped =
                        BaseVector::wrap_in_dictionary(nullptr, indices, dictionarySize, flat);
                EXPECT_EQ(wrapped->encoding(), VectorEncoding::Simple::DICTIONARY);

                wrapped = BaseVector::wrap_in_dictionary(
                    nullptr, indices, dictionarySize, flat, true /*flattenIfRedundant*/);
                EXPECT_EQ(wrapped->encoding(), VectorEncoding::Simple::DICTIONARY);
            }
        }

        TEST_F(
            VectorTest,
            verifyOuterDictionaryLayersInitalizedWhenLoggingLazyLoadedVector) {
            // Tester function verifies proper behavior when outer dictionary layers of a
            // lazy vector are not properly initalized and user calls toString. Expected
            // behavior should be a runtime error noting user to properly load vector
            // prior to invoking toString.
            auto doubleVector = make_flat_vector<double>({1.0, 2.0, 3.0});
            auto doubleInDictionary = BaseVector::wrap_in_dictionary(
                nullptr, make_indices({0, 1, 2}), 3, doubleVector);
            auto doubleInNestedDictionary = BaseVector::wrap_in_dictionary(
                nullptr, make_indices({0, 1, 2}), 3, doubleInDictionary);
            auto doubleInLazyDictionary =
                    VectorFuzzer::wrapInLazyVector(doubleInNestedDictionary);

            auto row_vector = make_row_vector({doubleVector, doubleInLazyDictionary});

            // Log vector; in doing so, we should trigger our POLLUX_CHECK error as
            // dictionary is not properly initialized.
            for (vector_size_t i = 0; i < row_vector->size(); ++i) {
                EXPECT_THROW(row_vector->toString(i), PolluxRuntimeError);
            }
        }

        TEST_F(VectorTest, hasOverlappingRanges) {
            auto test = [this](
                vector_size_t length,
                const std::vector<bool> &nulls,
                const std::vector<vector_size_t> &offsets,
                const std::vector<vector_size_t> &sizes,
                bool overlap) {
                auto makeArray = [&] {
                    return std::make_shared<ArrayVector>(
                        pool(),
                        ARRAY(BIGINT()),
                        make_nulls(nulls),
                        length,
                        make_indices(length, [&](auto i) { return offsets[i]; }),
                        make_indices(length, [&](auto i) { return sizes[i]; }),
                        make_null_constant(
                            TypeKind::BIGINT, std::numeric_limits<vector_size_t>::max()));
                };
                if (!overlap) {
                    ASSERT_FALSE(makeArray()->hasOverlappingRanges());
                } else {
                    ASSERT_TRUE(makeArray()->hasOverlappingRanges());
                }
            };
            test(3, {false, false, false}, {0, 1, 2}, {1, 1, 1}, false);
            test(3, {false, true, false}, {0, 0, 2}, {1, 1, 1}, false);
            test(3, {false, false, false}, {0, 0, 2}, {1, 0, 1}, false);
            test(3, {false, false, false}, {0, 0, 2}, {1, 1, 1}, true);
            test(3, {false, false, false}, {2, 1, 0}, {1, 1, 1}, false);
            test(3, {false, false, false}, {2, 1, 0}, {1, 2, 1}, true);
            test(2, {false, false}, {0, 1}, {3, 1}, true);
            test(2, {false, false}, {1, 0}, {1, 3}, true);
        }

        TEST_F(VectorTest, estimate_flat_size) {
            auto array_vector = make_array_vector<int64_t>(
                10, [](auto) { return 1; }, [](auto i, auto) { return i; });
            auto originalSize = array_vector->estimate_flat_size();
            array_vector->prepare_for_reuse();
            auto flatSize = array_vector->estimate_flat_size();
            EXPECT_NE(originalSize, flatSize);
            // Test that the second call to prepare_for_reuse will not cause crash
            array_vector->prepare_for_reuse();
        }
    } // namespace
} // namespace kumo::pollux
